text:
"""Compiler routines managing relation ranges and scope."""

from __future__ import annotations
from typing import *

import collections
import uuid

from edb import errors

from edb.edgeql import qltypes
from edb.edgeql import ast as qlast

from edb.ir import ast as irast
from edb.ir import typeutils as irtyputils
from edb.ir import utils as irutils

from edb.schema import pointers as s_pointers
from edb.schema import name as sn

from edb.pgsql import ast as pgast
from edb.pgsql import common
from edb.pgsql import types as pg_types

from . import astutils
from . import context
from . import dispatch
from . import output
from . import pathctx

def init_toplevel_query(
        ir_set: irast.Set, *,
        ctx: context.CompilerContextLevel) -> None:

    ctx.toplevel_stmt = ctx.stmt = ctx.rel
    update_scope(ir_set, ctx.rel, ctx=ctx)
    ctx.pending_query = ctx.rel

def _pull_path_namespace(
        *, target: pgast.Query, source: pgast.PathRangeVar,
        flavor: str='normal',
        replace_bonds: bool=True, ctx: context.CompilerContextLevel) -> None:

    squery = source.query
    source_qs: List[pgast.BaseRelation]

    if astutils.is_set_op_query(squery):
        # Set op query
        assert squery.larg and squery.rarg
        source_qs = [squery, squery.larg, squery.rarg]
    else:
        source_qs = [squery]

    for source_q in source_qs:
        s_paths: Set[Tuple[irast.PathId, str]] = set()
        if flavor == 'normal':
            if hasattr(source_q, 'path_outputs'):
                s_paths.update(source_q.path_outputs)
            if hasattr(source_q, 'path_namespace'):
                s_paths.update(source_q.path_namespace)
            if isinstance(source_q, pgast.Query):
                s_paths.update(source_q.path_rvar_map)
        elif flavor == 'packed':
            if hasattr(source_q, 'packed_path_outputs'):
                if source_q.packed_path_outputs:
                    s_paths.update(source_q.packed_path_outputs)
            if isinstance(source_q, pgast.Query):
                if source_q.path_packed_rvar_map:
                    s_paths.update(source_q.path_packed_rvar_map)
        else:
            raise AssertionError(f'unexpected flavor "{flavor}"')

        view_path_id_map = getattr(source_q, 'view_path_id_map', {})

        for path_id, aspect in s_paths:
            orig_path_id = path_id
            if flavor != 'packed':
                path_id = pathctx.reverse_map_path_id(
                    path_id, view_path_id_map)

            # Skip pulling paths that match the path_id_mask before or after
            # doing path id mapping. We need to look at before as well
            # to prevent paths leaking out under a different name.
            if flavor != 'packed' and (
                path_id in squery.path_id_mask
                or orig_path_id in squery.path_id_mask
            ):
                continue

            rvar = pathctx.maybe_get_path_rvar(
                target, path_id, aspect=aspect, flavor=flavor, env=ctx.env)
            if rvar is None or flavor == 'packed':
                pathctx.put_path_rvar(
                    target, path_id, source, aspect=aspect, flavor=flavor,
                    env=ctx.env)

def pull_path_namespace(
        *, target: pgast.Query, source: pgast.PathRangeVar,
        replace_bonds: bool=True, ctx: context.CompilerContextLevel) -> None:
    for flavor in ('normal', 'packed'):
        _pull_path_namespace(target=target, source=source, flavor=flavor,
                             replace_bonds=replace_bonds, ctx=ctx)

def find_rvar(
        stmt: pgast.Query, *,
        flavor: str='normal',
        source_stmt: Optional[pgast.Query]=None,
        path_id: irast.PathId,
        ctx: context.CompilerContextLevel) -> \
        Optional[pgast.PathRangeVar]:
    """Find an existing range var for a given *path_id* in stmt hierarchy.

    If a range var is visible in a given SQL scope denoted by *stmt*, or,
    optionally, *source_stmt*, record it on *stmt* for future reference.

    :param stmt:
        The statement to ensure range var visibility in.

    :param flavor:
        Whether to look for normal rvars or packed rvars

    :param source_stmt:
        An optional statement object which is used as the starting SQL scope
        for range var search.  If not specified, *stmt* is used as the
        starting scope.

    :param path_id:
        The path ID of the range var being searched.

    :param ctx:
        Compiler context.

    :return:
        A range var instance if found, ``None`` otherwise.
    """

    if source_stmt is None:
        source_stmt = stmt

    rvar = maybe_get_path_rvar(source_stmt, path_id=path_id,
                               aspect='value', flavor=flavor, ctx=ctx)
    if rvar is not None:
        pathctx.put_path_rvar_if_not_exists(
            stmt, path_id, rvar, aspect='value', flavor=flavor, env=ctx.env)

        src_rvar = maybe_get_path_rvar(source_stmt, path_id=path_id,
                                       aspect='source', flavor=flavor, ctx=ctx)

        if src_rvar is not None:
            pathctx.put_path_rvar_if_not_exists(
                stmt, path_id, src_rvar,
                aspect='source', flavor=flavor, env=ctx.env)

    return rvar

def include_rvar(
        stmt: pgast.SelectStmt,
        rvar: pgast.PathRangeVar,
        path_id: irast.PathId, *,
        overwrite_path_rvar: bool=False,
        pull_namespace: bool=True,
        update_mask: bool=True,
        flavor: str='normal',
        aspects: Optional[Tuple[str, ...] | AbstractSet[str]]=None,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    """Ensure that *rvar* is visible in *stmt* as a value/source aspect.

    :param stmt:
        The statement to include *rel* in.

    :param rvar:
        The range var node to join.

    :param join_type:
        JOIN type to use when including *rel*.

    :param flavor:
        Whether this is a normal or packed rvar

    :param aspect:
        The reference aspect of the range var.

    :param ctx:
        Compiler context.
    """
    if aspects is None:
        aspects = ('value',)
        if path_id.is_objtype_path():
            if isinstance(rvar, pgast.RangeSubselect):
                if pathctx.has_path_aspect(
                        rvar.query, path_id, aspect='source', env=ctx.env):
                    aspects += ('source',)
            else:
                aspects += ('source',)

    return include_specific_rvar(
        stmt, rvar=rvar, path_id=path_id,
        overwrite_path_rvar=overwrite_path_rvar,
        pull_namespace=pull_namespace,
        update_mask=update_mask,
        flavor=flavor,
        aspects=aspects,
        ctx=ctx)

def include_specific_rvar(
        stmt: pgast.SelectStmt,
        rvar: pgast.PathRangeVar,
        path_id: irast.PathId, *,
        overwrite_path_rvar: bool=False,
        pull_namespace: bool=True,
        update_mask: bool=True,
        flavor: str='normal',
        aspects: Iterable[str]=('value',),
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    """Make the *aspect* of *path_id* visible in *stmt* as *rvar*.

    :param stmt:
        The statement to include *rel* in.

    :param rvar:
        The range var node to join.

    :param join_type:
        JOIN type to use when including *rel*.

    :param flavor:
        Whether this is a normal or packed rvar

    :param aspect:
        The reference aspect of the range var.

    :param ctx:
        Compiler context.
    """

    if not has_rvar(stmt, rvar, ctx=ctx):
        if not (
            ctx.env.external_rvars
            and has_external_rvar(rvar, ctx=ctx)
        ):
            rel_join(stmt, rvar, ctx=ctx)
        # Make sure that the path namespace of *rvar* is mapped
        # onto the path namespace of *stmt*.
        if pull_namespace:
            pull_path_namespace(target=stmt, source=rvar, ctx=ctx)

    for aspect in aspects:
        if overwrite_path_rvar:
            pathctx.put_path_rvar(
                stmt, path_id, rvar, flavor=flavor, aspect=aspect, env=ctx.env)
        else:
            pathctx.put_path_rvar_if_not_exists(
                stmt, path_id, rvar, flavor=flavor, aspect=aspect, env=ctx.env)

    if update_mask:
        scopes = [ctx.scope_tree]
        parent_scope = ctx.scope_tree.parent
        if parent_scope is not None:
            scopes.append(parent_scope)

        tpath_id = path_id.tgt_path()
        if not any(scope.path_id == tpath_id or
                   scope.find_child(tpath_id) for scope in scopes):
            pathctx.put_path_id_mask(stmt, path_id)

    return rvar

def has_rvar(
        stmt: pgast.Query, rvar: pgast.PathRangeVar, *,
        ctx: context.CompilerContextLevel) -> bool:

    curstmt: Optional[pgast.Query] = stmt

    while curstmt is not None:
        if pathctx.has_rvar(curstmt, rvar, env=ctx.env):
            return True
        curstmt = ctx.rel_hierarchy.get(curstmt)

    return False

def has_external_rvar(
    rvar: pgast.PathRangeVar,
    *,
    ctx: context.CompilerContextLevel,
) -> bool:
    return rvar in ctx.env.external_rvars.values()

def _maybe_get_path_rvar(
    stmt: pgast.Query,
    path_id: irast.PathId,
    *,
    flavor: str='normal',
    aspect: str,
    ctx: context.CompilerContextLevel,
) -> Optional[Tuple[pgast.PathRangeVar, irast.PathId]]:
    qry: Optional[pgast.Query] = stmt
    while qry is not None:
        rvar = pathctx.maybe_get_path_rvar(
            qry, path_id, aspect=aspect, flavor=flavor, env=ctx.env)
        if rvar is not None:
            if qry is not stmt:
                # Cache the rvar reference.
                pathctx.put_path_rvar(stmt, path_id, rvar,
                                      flavor=flavor, aspect=aspect,
                                      env=ctx.env)
            return rvar, path_id
        qry = ctx.rel_hierarchy.get(qry)

    return None

def _get_path_rvar(
    stmt: pgast.Query,
    path_id: irast.PathId,
    *,
    flavor: str='normal',
    aspect: str,
    ctx: context.CompilerContextLevel,
) -> Tuple[pgast.PathRangeVar, irast.PathId]:
    result = _maybe_get_path_rvar(
        stmt, path_id, flavor=flavor, aspect=aspect, ctx=ctx)
    if result is None:
        raise LookupError(f'there is no range var for {path_id} in {stmt}')
    else:
        return result

def get_path_rvar(
        stmt: pgast.Query, path_id: irast.PathId, *,
        flavor: str='normal',
        aspect: str, ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    return _get_path_rvar(
        stmt, path_id, flavor=flavor, aspect=aspect, ctx=ctx)[0]

def get_path_var(
        stmt: pgast.Query, path_id: irast.PathId, *,
        aspect: str, ctx: context.CompilerContextLevel) -> pgast.BaseExpr:
    var = pathctx.maybe_get_path_var(
        stmt, path_id=path_id, aspect=aspect, env=ctx.env)
    if var is not None:
        return var
    else:
        rvar, path_id = _get_path_rvar(stmt, path_id, aspect=aspect, ctx=ctx)
        return pathctx.get_rvar_path_var(
            rvar, path_id, aspect=aspect, env=ctx.env)

def maybe_get_path_rvar(
        stmt: pgast.Query, path_id: irast.PathId, *,
        flavor: str='normal',
        aspect: str, ctx: context.CompilerContextLevel
) -> Optional[pgast.PathRangeVar]:
    result = _maybe_get_path_rvar(stmt, path_id,
                                  aspect=aspect, flavor=flavor, ctx=ctx)
    return result[0] if result is not None else None

def maybe_get_path_var(
        stmt: pgast.Query, path_id: irast.PathId, *,
        aspect: str, ctx: context.CompilerContextLevel
) -> Optional[pgast.BaseExpr]:
    result = _maybe_get_path_rvar(stmt, path_id, aspect=aspect, ctx=ctx)
    if result is None:
        return None
    else:
        try:
            return pathctx.get_rvar_path_var(
                result[0], result[1], aspect=aspect, env=ctx.env)
        except LookupError:
            return None

def new_empty_rvar(
        ir_set: irast.EmptySet, *,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    nullrel = pgast.NullRelation(path_id=ir_set.path_id)
    rvar = rvar_for_rel(nullrel, ctx=ctx)
    pathctx.put_rvar_path_bond(rvar, ir_set.path_id)
    return rvar

def new_free_object_rvar(
    typeref: irast.TypeRef,
    path_id: irast.PathId,
    *,
    lateral: bool=False,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:
    """Create a fake source rel for a free object

    We generate fake IDs for free objects. The only thing other than ids
    that need to come from a free object is __type__, which we inject
    in a special case way in pathctx.get_path_var.

    We also have a special case in relgen.ensure_source_rvar to reuse an
    existing value rvar instead of creating a new root rvar.

    (We inject __type__ in get_path_var instead of injecting it here because
    we don't have the pathid for it available to us here and because it
    allows ensure_source_rvar to simply reuse a value rvar.)

    """
    with ctx.subrel() as subctx:
        qry = subctx.rel

        id_expr = pgast.FuncCall(
            name=('edgedbext', 'uuid_generate_v4',), args=[]
        )

        pathctx.put_path_identity_var(qry, path_id, id_expr, env=ctx.env)
        pathctx.put_path_value_var(qry, path_id, id_expr, env=ctx.env)
        apply_volatility_ref(qry, ctx=subctx)

    return rvar_for_rel(qry, typeref=typeref, lateral=lateral, ctx=ctx)

def _deep_copy_primitive_rvar_path_var(
    orig_id: irast.PathId, new_id: irast.PathId,
    rvar: pgast.PathRangeVar, *,
    env: context.Environment
) -> None:
    """Copy one identity path to another in a primitive rvar.

    The trickiness here is because primitive rvars might have an
    overlay stack, which means if they are joined on, it might be
    using _lateral_union_join, which requires every component of
    the union to have all the path bonds.
    """

    def _copy_id(component: pgast.Query) -> None:
        rref = pathctx.get_path_var(
            component, orig_id, aspect='identity', env=env)
        pathctx.put_path_var(
            component, new_id, rref, aspect='identity',
            env=env)

    if isinstance(rvar, pgast.RangeSubselect):
        astutils.for_each_query_in_set(rvar.query, _copy_id)
    else:
        rref = pathctx.get_path_output(
            rvar.query, orig_id, aspect='identity', env=env)
        pathctx.put_rvar_path_output(
            rvar, new_id, aspect='identity', var=rref, env=env)

def new_primitive_rvar(
    ir_set: irast.Set,
    *,
    path_id: irast.PathId,
    lateral: bool,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:
    if not ir_set.path_id.is_objtype_path():
        raise ValueError('cannot create root rvar for non-object path')

    typeref = ir_set.typeref
    dml_source = irutils.get_nearest_dml_stmt(ir_set)
    set_rvar = range_for_typeref(
        typeref, path_id, lateral=lateral, dml_source=dml_source,
        include_descendants=not ir_set.skip_subtypes,
        ignore_rewrites=ir_set.ignore_rewrites, ctx=ctx)
    pathctx.put_rvar_path_bond(set_rvar, path_id)

    rptr = ir_set.rptr
    if rptr is not None:
        if (isinstance(rptr.ptrref, irast.TypeIntersectionPointerRef)
                and rptr.source.rptr):
            rptr = rptr.source.rptr

        # If the set comes from an backlink, and the link is stored inline,
        # we want to output the source path.
        if (
            rptr.is_inbound
            and (
                rptrref := irtyputils.maybe_find_actual_ptrref(
                    set_rvar.typeref, rptr.ptrref) or rptr.ptrref
                if set_rvar.typeref else rptr.ptrref
            ) and (
                ptr_info := pg_types.get_ptrref_storage_info(
                    rptrref, resolve_type=False, link_bias=False,
                    allow_missing=True)
            ) and ptr_info.table_type == 'ObjectType'
        ):
            # Inline link
            prefix_path_id = path_id.src_path()
            assert prefix_path_id is not None, 'expected a path'

            flipped_id = path_id.extend(ptrref=rptrref)

            # Unfortunately we can't necessarily just install the
            # prefix path id path---the rvar from range_from_typeref
            # might be a DML overlay, which means joins on it will try
            # to use _lateral_union_join; this means that all of the
            # path bonds need to be valid on each *subquery*, so we
            # need to set them up in each subquery.
            _deep_copy_primitive_rvar_path_var(
                flipped_id, prefix_path_id, set_rvar, env=ctx.env)
            pathctx.put_rvar_path_bond(set_rvar, prefix_path_id)

    return set_rvar

def new_root_rvar(
    ir_set: irast.Set,
    *,
    lateral: bool = False,
    path_id: Optional[irast.PathId] = None,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:

    if path_id is None:
        path_id = ir_set.path_id

    if irtyputils.is_free_object(path_id.target):
        return new_free_object_rvar(
            path_id.target, path_id, lateral=lateral, ctx=ctx)

    narrowing = ctx.intersection_narrowing.get(ir_set)
    if narrowing is not None:
        ir_set = narrowing
    return new_primitive_rvar(
        ir_set, lateral=lateral, path_id=path_id, ctx=ctx)

def new_pointer_rvar(
        ir_ptr: irast.Pointer, *,
        link_bias: bool=False,
        src_rvar: pgast.PathRangeVar,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:

    ptrref = ir_ptr.ptrref

    ptr_info = pg_types.get_ptrref_storage_info(
        ptrref, resolve_type=False, link_bias=link_bias, allow_missing=True)

    if ptr_info and ptr_info.table_type == 'ObjectType':
        # Inline link
        return _new_inline_pointer_rvar(
            ir_ptr, ptr_info=ptr_info,
            src_rvar=src_rvar, ctx=ctx)
    else:
        return _new_mapped_pointer_rvar(ir_ptr, ctx=ctx)

def _new_inline_pointer_rvar(
        ir_ptr: irast.Pointer, *,
        lateral: bool=True,
        ptr_info: pg_types.PointerStorageInfo,
        src_rvar: pgast.PathRangeVar,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    ptr_rel = pgast.SelectStmt()
    ptr_rvar = rvar_for_rel(ptr_rel, lateral=lateral, ctx=ctx)
    ptr_rvar.query.path_id = ir_ptr.target.path_id.ptr_path()

    is_inbound = ir_ptr.direction == s_pointers.PointerDirection.Inbound

    if is_inbound:
        far_pid = ir_ptr.source.path_id
    else:
        far_pid = ir_ptr.target.path_id

    far_ref = pathctx.get_rvar_path_identity_var(
        src_rvar, far_pid, env=ctx.env)

    pathctx.put_rvar_path_bond(ptr_rvar, far_pid)
    pathctx.put_path_identity_var(ptr_rel, far_pid, var=far_ref, env=ctx.env)

    return ptr_rvar

def _new_mapped_pointer_rvar(
        ir_ptr: irast.Pointer, *,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    ptrref = ir_ptr.ptrref
    dml_source = irutils.get_nearest_dml_stmt(ir_ptr.source)
    ptr_rvar = range_for_pointer(ir_ptr, dml_source=dml_source, ctx=ctx)

    src_col = 'source'
    source_ref = pgast.ColumnRef(name=[src_col], nullable=False)

    tgt_col = 'target'
    target_ref = pgast.ColumnRef(
        name=[tgt_col],
        nullable=not ptrref.required)

    # Set up references according to the link direction.
    if (
        ir_ptr.direction == s_pointers.PointerDirection.Inbound
        or ptrref.computed_backlink
    ):
        near_ref = target_ref
        far_ref = source_ref
    else:
        near_ref = source_ref
        far_ref = target_ref

    src_pid = ir_ptr.source.path_id
    tgt_pid = ir_ptr.target.path_id
    ptr_pid = tgt_pid.ptr_path()

    ptr_rvar.query.path_id = ptr_pid
    pathctx.put_rvar_path_bond(ptr_rvar, src_pid)
    pathctx.put_rvar_path_output(ptr_rvar, src_pid, aspect='identity',
                                 var=near_ref, env=ctx.env)
    pathctx.put_rvar_path_output(ptr_rvar, src_pid, aspect='value',
                                 var=near_ref, env=ctx.env)
    pathctx.put_rvar_path_output(ptr_rvar, tgt_pid, aspect='value',
                                 var=far_ref, env=ctx.env)

    if tgt_pid.is_objtype_path():
        pathctx.put_rvar_path_bond(ptr_rvar, tgt_pid)
        pathctx.put_rvar_path_output(ptr_rvar, tgt_pid, aspect='identity',
                                     var=far_ref, env=ctx.env)

    return ptr_rvar

def is_pointer_rvar(
    rvar: pgast.PathRangeVar,
    *,
    ctx: context.CompilerContextLevel,
) -> bool:
    return rvar.query.path_id is not None and rvar.query.path_id.is_ptr_path()

def new_rel_rvar(
        ir_set: irast.Set, stmt: pgast.Query, *,
        lateral: bool=True,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    return rvar_for_rel(stmt, typeref=ir_set.typeref, lateral=lateral, ctx=ctx)

def semi_join(
        stmt: pgast.SelectStmt,
        ir_set: irast.Set, src_rvar: pgast.PathRangeVar, *,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    """Join an IR Set using semi-join."""
    rptr = ir_set.rptr
    assert rptr is not None

    # Target set range.
    set_rvar = new_root_rvar(ir_set, lateral=True, ctx=ctx)

    ptrref = rptr.ptrref
    ptr_info = pg_types.get_ptrref_storage_info(
        ptrref, resolve_type=False, allow_missing=True)

    if ptr_info and ptr_info.table_type == 'ObjectType':
        if rptr.is_inbound:
            far_pid = ir_set.path_id.src_path()
            assert far_pid is not None
        else:
            far_pid = ir_set.path_id
    else:
        far_pid = ir_set.path_id
        # Link range.
        map_rvar = new_pointer_rvar(rptr, src_rvar=src_rvar, ctx=ctx)
        include_rvar(
            ctx.rel, map_rvar,
            path_id=ir_set.path_id.ptr_path(), ctx=ctx)

    tgt_ref = pathctx.get_rvar_path_identity_var(
        set_rvar, far_pid, env=ctx.env)

    pathctx.get_path_identity_output(
        ctx.rel, far_pid, env=ctx.env)

    cond = astutils.new_binop(tgt_ref, ctx.rel, 'IN')
    stmt.where_clause = astutils.extend_binop(
        stmt.where_clause, cond)

    return set_rvar

def ensure_bond_for_expr(
    ir_set: irast.Set,
    stmt: pgast.BaseRelation,
    *,
    ctx: context.CompilerContextLevel,
) -> None:
    if ir_set.path_id.is_objtype_path():
        # ObjectTypes have inherent identity
        return

    ensure_transient_identity_for_path(ir_set.path_id, stmt, ctx=ctx)

def apply_volatility_ref(
        stmt: pgast.SelectStmt, *,
        ctx: context.CompilerContextLevel) -> None:
    for ref in ctx.volatility_ref:
        # Apply the volatility reference.
        # See the comment in process_set_as_subquery().
        arg = ref(stmt, ctx)
        if not arg:
            continue
        stmt.where_clause = astutils.extend_binop(
            stmt.where_clause,
            pgast.NullTest(
                arg=arg,
                negated=True,
            )
        )

def ensure_transient_identity_for_path(
    path_id: irast.PathId,
    stmt: pgast.BaseRelation,
    *,
    ctx: context.CompilerContextLevel,
) -> None:

    id_expr = pgast.FuncCall(
        name=('edgedbext', 'uuid_generate_v4',),
        args=[],
    )

    pathctx.put_path_identity_var(
        stmt, path_id, id_expr, force=True, env=ctx.env)
    pathctx.put_path_bond(stmt, path_id)

    if isinstance(stmt, pgast.SelectStmt):
        apply_volatility_ref(stmt, ctx=ctx)

def get_scope(
    ir_set: irast.Set, *,
    ctx: context.CompilerContextLevel,
) -> Optional[irast.ScopeTreeNode]:

    result: Optional[irast.ScopeTreeNode] = None

    if ir_set.path_scope_id is not None:
        result = ctx.env.scope_tree_nodes.get(ir_set.path_scope_id)

    return result

def update_scope(
        ir_set: irast.Set, stmt: pgast.SelectStmt, *,
        ctx: context.CompilerContextLevel) -> None:
    """Update the scope of an ir set to be a pg stmt.

    If ir_set has a scope node associated with it, update path_scope
    so that any paths bound in that scope will be compiled in the
    context of stmt.

    This, combined with maybe_get_scope_stmt, is the mechanism by
    which the scope tree influences the shape of the output query.
    """

    scope_tree = get_scope(ir_set, ctx=ctx)
    if scope_tree is None:
        return

    ctx.scope_tree = scope_tree
    ctx.path_scope = ctx.path_scope.new_child()

    # Register paths in the current scope to be compiled as a subrel
    # of stmt.
    for p in scope_tree.path_children:
        assert p.path_id is not None
        ctx.path_scope[p.path_id] = stmt

    # Mark any paths under the scope tree as masked, so that they
    # won't get picked up by pull_path_namespace.
    for child_path in scope_tree.get_all_paths():
        pathctx.put_path_id_mask(stmt, child_path)

def maybe_get_scope_stmt(
    path_id: irast.PathId,
    *,
    ctx: context.CompilerContextLevel,
) -> Optional[pgast.SelectStmt]:
    stmt = ctx.path_scope.get(path_id)
    if stmt is None and path_id.is_ptr_path():
        stmt = ctx.path_scope.get(path_id.tgt_path())
    return stmt

def set_to_array(
        path_id: irast.PathId, query: pgast.Query, *,
        for_group_by: bool=False,
        ctx: context.CompilerContextLevel) -> pgast.Query:
    """Collapse a set into an array."""
    subrvar = pgast.RangeSubselect(
        subquery=query,
        alias=pgast.Alias(
            aliasname=ctx.env.aliases.get('aggw')
        )
    )

    result = pgast.SelectStmt()
    aspects = pathctx.list_path_aspects(subrvar.query, path_id, env=ctx.env)
    include_rvar(result, subrvar, path_id=path_id, aspects=aspects, ctx=ctx)

    val: Optional[pgast.BaseExpr] = (
        pathctx.maybe_get_path_serialized_var(
            result, path_id, env=ctx.env)
    )

    if val is None:
        value_var = pathctx.get_path_value_var(
            result, path_id, env=ctx.env)
        val = output.serialize_expr(
            value_var, path_id=path_id, env=ctx.env)
        pathctx.put_path_serialized_var(
            result, path_id, val, force=True, env=ctx.env)

    if isinstance(val, pgast.TupleVarBase):
        val = output.serialize_expr(
            val, path_id=path_id, env=ctx.env)

    pg_type = output.get_pg_type(path_id.target, ctx=ctx)

    agg_filter_safe = True

    if for_group_by:
        # When doing this as part of a GROUP, the stuff being aggregated
        # needs to actually appear *inside* of the aggregate call...
        result.target_list = [pgast.ResTarget(val=val, ser_safe=val.ser_safe)]
        val = result
        try_collapse = astutils.collapse_query(val)
        if isinstance(try_collapse, pgast.ColumnRef):
            val = try_collapse
        else:
            agg_filter_safe = False

        result = pgast.SelectStmt()

    orig_val = val

    if (path_id.is_array_path()
            and ctx.env.output_format is context.OutputFormat.NATIVE):
        # We cannot aggregate arrays straight away, as
        # they be of different length, so we have to
        # encase each element into a record.
        val = pgast.RowExpr(args=[val], ser_safe=val.ser_safe)
        pg_type = ('record',)

    array_agg = pgast.FuncCall(
        name=('array_agg',),
        args=[val],
        agg_filter=(
            astutils.new_binop(orig_val, pgast.NullConstant(),
                               'IS DISTINCT FROM')
            if orig_val.nullable and agg_filter_safe else None
        ),
        ser_safe=val.ser_safe,
    )

    # If this is for a group by, and the body isn't just a column ref,
    # then we need to remove NULLs after the fact.
    if orig_val.nullable and not agg_filter_safe:
        array_agg = pgast.FuncCall(
            name=('array_remove',),
            args=[array_agg, pgast.NullConstant()]
        )

    agg_expr = pgast.CoalesceExpr(
        args=[
            array_agg,
            pgast.TypeCast(
                arg=pgast.ArrayExpr(elements=[]),
                type_name=pgast.TypeName(name=pg_type, array_bounds=[-1])
            )
        ],
        ser_safe=array_agg.ser_safe,
        nullable=False,
    )

    result.target_list = [
        pgast.ResTarget(
            name=ctx.env.aliases.get('v'),
            val=agg_expr,
            ser_safe=agg_expr.ser_safe,
        )
    ]

    return result

class UnpackElement(NamedTuple):
    path_id: irast.PathId
    colname: str
    packed: bool
    multi: bool
    ref: Optional[pgast.BaseExpr]

def unpack_rvar(
        stmt: pgast.SelectStmt, path_id: irast.PathId, *,
        packed_rvar: pgast.PathRangeVar,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:
    ref = pathctx.get_rvar_path_var(
        packed_rvar, path_id, aspect='value', flavor='packed', env=ctx.env)
    return unpack_var(stmt, path_id, ref=ref, ctx=ctx)

def unpack_var(
        stmt: pgast.SelectStmt, path_id: irast.PathId, *,
        ref: pgast.OutputVar,
        ctx: context.CompilerContextLevel) -> pgast.PathRangeVar:

    qry = pgast.SelectStmt()

    view_tvars: List[Tuple[irast.PathId, pgast.TupleVarBase, bool]] = []
    els = []
    ctr = 0

    def walk(ref: pgast.BaseExpr, path_id: irast.PathId, multi: bool) -> None:
        nonlocal ctr

        coldeflist = []
        alias = ctx.env.aliases.get('unpack')
        simple = False
        if irtyputils.is_tuple(path_id.target):
            els.append(UnpackElement(
                path_id, alias, packed=False, multi=False, ref=None
            ))

            orig_view_count = len(view_tvars)
            tuple_tvar_elements = []
            for i, st in enumerate(path_id.target.subtypes):
                colname = f'_t{ctr}'
                ctr += 1

                typ = pg_types.pg_type_from_ir_typeref(st)
                if st.id in ctx.env.materialized_views:
                    typ = ('record',)

                # Construct a path_id for the element
                el_name = sn.QualName('__tuple__', st.element_name or str(i))
                el_ref = irast.TupleIndirectionPointerRef(
                    name=el_name, shortname=el_name,
                    out_source=path_id.target,
                    out_target=st,
                    out_cardinality=qltypes.Cardinality.ONE,
                )
                el_path_id = path_id.extend(ptrref=el_ref)

                el_var = (
                    astutils.tuple_getattr(
                        pgast.ColumnRef(name=[alias]),
                        path_id.target, el_name.name)
                    if irtyputils.is_persistent_tuple(path_id.target)
                    else pgast.ColumnRef(name=[colname])
                )
                walk(el_var, el_path_id, multi=False)

                tuple_tvar_elements.append(
                    pgast.TupleElementBase(
                        path_id=el_path_id, name=el_name.name
                    )
                )

                coldeflist.append(
                    pgast.ColumnDef(
                        name=colname,
                        typename=pgast.TypeName(name=typ)
                    )
                )

            if len(view_tvars) > orig_view_count:
                tuple_tvar = pgast.TupleVarBase(
                    elements=tuple_tvar_elements,
                    typeref=path_id.target,
                    named=any(
                        st.element_name for st in path_id.target.subtypes),
                )
                view_tvars.append((path_id, tuple_tvar, True))

            if irtyputils.is_persistent_tuple(path_id.target):
                coldeflist = []

        elif irtyputils.is_array(path_id.target) and multi:
            # TODO: materialized arrays of tuples and arrays are really
            # quite broken
            coldeflist = [
                pgast.ColumnDef(
                    name='q',
                    typename=pgast.TypeName(
                        name=pg_types.pg_type_from_ir_typeref(
                            path_id.target)
                    )
                )
            ]

            els.append(UnpackElement(
                path_id, coldeflist[0].name,
                packed=False, multi=False, ref=None
            ))

        elif path_id.target.id in ctx.env.materialized_views:
            view_tuple = ctx.env.materialized_views[path_id.target.id]

            vpath_ids = []
            id_idx = None
            for el, _ in view_tuple.shape:
                src_path, el_ptrref = el.path_id.src_path(), el.path_id.rptr()
                assert src_path and el_ptrref

                el_id = path_id.ptr_path().extend(ptrref=el_ptrref)

                assert el.rptr
                card = el.rptr.ptrref.dir_cardinality(el.rptr.direction)
                is_singleton = card.is_single() and not card.can_be_zero()
                must_pack = not is_singleton

                if (rptr_name := el_id.rptr_name()) and rptr_name.name == 'id':
                    id_idx = len(els)

                colname = f'_t{ctr}'
                ctr += 1

                typ = pg_types.pg_type_from_ir_typeref(el_id.target)
                if el_id.target.id in ctx.env.materialized_views:
                    typ = ('record',)
                    must_pack = True

                if not is_singleton:
                    # Arrays get wrapped in a record before they can be put
                    # in another array
                    if el_id.is_array_path():
                        typ = ('record',)
                        must_pack = True
                    typ = pg_types.pg_type_array(typ)

                coldeflist.append(
                    pgast.ColumnDef(
                        name=colname,
                        typename=pgast.TypeName(name=typ),
                    )
                )
                els.append(UnpackElement(
                    el_id, colname,
                    packed=must_pack, multi=not is_singleton, ref=None
                ))

                vpath_ids.append(el_id)

            if id_idx is not None:
                els.append(UnpackElement(
                    path_id, els[id_idx].colname,
                    multi=False, packed=False, ref=None,
                ))

            view_tvars.append((path_id, pgast.TupleVarBase(
                elements=[
                    pgast.TupleElementBase(
                        path_id=pid,
                        name=astutils.tuple_element_for_shape_el(
                            el, ctx=ctx).name,
                    )
                    for (el, op), pid in zip(view_tuple.shape, vpath_ids)
                    if op != qlast.ShapeOp.MATERIALIZE or ctx.materializing
                ],
                typeref=path_id.target,
                named=True,
            ), False))

        else:
            coldeflist = []
            simple = not multi
            els.append(UnpackElement(
                path_id, alias, multi=False, packed=False,
                ref=ref if simple else None,
            ))

        if not simple:
            if not multi:
                # Sigh, have to wrap in an array so we can unpack.
                ref = pgast.ArrayExpr(elements=[ref])

            qry.from_clause.insert(
                0,
                pgast.RangeFunction(
                    alias=pgast.Alias(
                        aliasname=alias,
                    ),
                    is_rowsfrom=True,
                    functions=[
                        pgast.FuncCall(
                            name=('unnest',),
                            args=[ref],
                            coldeflist=coldeflist,
                        )
                    ]
                )
            )

    ########################

    walk(ref, path_id, ref.is_packed_multi)

    rvar = rvar_for_rel(qry, lateral=True, ctx=ctx)
    include_rvar(stmt, rvar, path_id=path_id, aspects=('value',), ctx=ctx)

    for el in els:
        el_id = el.path_id
        cur_ref = el.ref or pgast.ColumnRef(name=[el.colname])

        for aspect in ('value', 'serialized'):
            pathctx.put_path_var(
                qry, el_id, cur_ref, aspect=aspect, env=ctx.env,
            )

        if not el.packed:
            pathctx.put_path_rvar(
                stmt, el_id, rvar, aspect='value', env=ctx.env)

            pathctx.put_path_rvar(
                ctx.rel, el_id, rvar, aspect='value', env=ctx.env)
        else:
            cref = pathctx.get_path_output(
                qry, el_id, aspect='value', env=ctx.env)
            cref = cref.replace(is_packed_multi=el.multi)

            pathctx.put_path_packed_output(qry, el_id, val=cref)

            pathctx.put_path_rvar(
                stmt, el_id, rvar, flavor='packed', aspect='value', env=ctx.env
            )

    # When we're producing an exposed shape, we need to rewrite the
    # serialized shape.
    # We also need to rewrite tuples that contain such shapes!
    # What a pain!
    #
    # We *also* need to rewrite tuple values, so that we don't consider
    # serialized materialized objects as part of the value of the tuple
    for view_path_id, view_tvar, is_tuple in view_tvars:
        if not view_tvar.elements:
            continue

        rewrite_aspects = []
        if ctx.expr_exposed and not is_tuple:
            rewrite_aspects.append('serialized')
        if is_tuple:
            rewrite_aspects.append('value')

        # Reserialize links if we are producing final output
        if (
            ctx.expr_exposed and not ctx.materializing and not is_tuple
        ):
            for tel in view_tvar.elements:
                el = [x for x in els if x.path_id == tel.path_id][0]
                if not el.packed:
                    continue
                reqry = reserialize_object(el, tel, ctx=ctx)
                pathctx.put_path_var(
                    qry, tel.path_id, reqry, aspect='serialized',
                    env=ctx.env, force=True
                )

        for aspect in rewrite_aspects:
            tv = pathctx.fix_tuple(qry, view_tvar, aspect=aspect, env=ctx.env)
            sval = (
                output.output_as_value(tv, env=ctx.env)
                if aspect == 'value' else
                output.serialize_expr(tv, path_id=path_id, env=ctx.env)
            )
            pathctx.put_path_var(
                qry, view_path_id, sval, aspect=aspect, env=ctx.env, force=True
            )
            pathctx.put_path_rvar(
                ctx.rel, view_path_id, rvar, aspect=aspect, env=ctx.env
            )

    return rvar

def reserialize_object(
        el: UnpackElement, tel: pgast.TupleElementBase,
        *,
        ctx: context.CompilerContextLevel) -> pgast.Query:
    tref = pgast.ColumnRef(name=[el.colname], is_packed_multi=el.multi)

    with ctx.subrel() as subctx:
        sub_rvar = unpack_var(subctx.rel, tel.path_id, ref=tref, ctx=subctx)
    reqry = sub_rvar.query
    assert isinstance(reqry, pgast.Query)
    rptr = tel.path_id.rptr()
    pathctx.get_path_serialized_output(reqry, tel.path_id, env=ctx.env)
    assert rptr
    if rptr.out_cardinality.is_multi():
        with ctx.subrel() as subctx:
            reqry = set_to_array(
                path_id=tel.path_id, query=reqry, ctx=subctx)
    return reqry

def get_scope_stmt(
    path_id: irast.PathId,
    *,
    ctx: context.CompilerContextLevel,
) -> pgast.SelectStmt:
    stmt = maybe_get_scope_stmt(path_id, ctx=ctx)
    if stmt is None:
        raise LookupError(f'cannot find scope statement for {path_id}')
    else:
        return stmt

def rel_join(
    query: pgast.SelectStmt,
    right_rvar: pgast.PathRangeVar,
    *,
    ctx: context.CompilerContextLevel,
) -> None:

    if (
        isinstance(right_rvar, pgast.RangeSubselect)
        and astutils.is_set_op_query(right_rvar.subquery)
        and right_rvar.tag == "overlay-stack"
        and all(astutils.for_each_query_in_set(
            right_rvar.subquery, lambda q: isinstance(q, pgast.SelectStmt)))
        and not is_pointer_rvar(right_rvar, ctx=ctx)
    ):
        # Unfortunately Postgres sometimes produces a very bad plan
        # when we join a UNION which is not a trivial Append, most notably
        # those produced by DML overlays.  To work around this we push
        # the JOIN condition into the WHERE clause of each UNION component.
        # While this is likely not harmful (and possibly beneficial) for
        # all kinds of UNIONs, we restrict this optimization to overlay
        # UNIONs only to limit the possibility of breakage as not all
        # UNIONs are guaranteed to have correct path namespace and
        # translation maps set up.
        _lateral_union_join(query, right_rvar, ctx=ctx)
    else:
        _plain_join(query, right_rvar, ctx=ctx)

def _plain_join(
    query: pgast.SelectStmt,
    right_rvar: pgast.PathRangeVar,
    *,
    ctx: context.CompilerContextLevel,
) -> None:
    condition = None

    for path_id in right_rvar.query.path_scope:
        lref = maybe_get_path_var(query, path_id, aspect='identity', ctx=ctx)
        if lref is None:
            lref = maybe_get_path_var(query, path_id, aspect='value', ctx=ctx)
        if lref is None:
            continue

        rref = pathctx.get_rvar_path_identity_var(
            right_rvar, path_id, env=ctx.env)

        assert isinstance(lref, pgast.ColumnRef)
        assert isinstance(rref, pgast.ColumnRef)
        path_cond = astutils.join_condition(lref, rref)
        condition = astutils.extend_binop(condition, path_cond)

    if condition is None:
        join_type = 'cross'
    else:
        join_type = 'inner'

    if not query.from_clause:
        query.from_clause.append(right_rvar)
        if condition is not None:
            query.where_clause = astutils.extend_binop(
                query.where_clause, condition)
    else:
        larg = query.from_clause[0]
        rarg = right_rvar

        query.from_clause[0] = pgast.JoinExpr(
            type=join_type, larg=larg, rarg=rarg, quals=condition)

def _lateral_union_join(
    query: pgast.SelectStmt,
    right_rvar: pgast.RangeSubselect,
    *,
    ctx: context.CompilerContextLevel,
) -> None:
    def _inject_filter(component: pgast.Query) -> None:
        condition = None

        for path_id in right_rvar.query.path_scope:
            lref = maybe_get_path_var(
                query, path_id, aspect='identity', ctx=ctx)
            if lref is None:
                lref = maybe_get_path_var(
                    query, path_id, aspect='value', ctx=ctx)
            if lref is None:
                continue

            rref = pathctx.get_path_identity_var(
                component, path_id, env=ctx.env)

            assert isinstance(lref, pgast.ColumnRef)
            assert isinstance(rref, pgast.ColumnRef)
            path_cond = astutils.join_condition(lref, rref)
            condition = astutils.extend_binop(condition, path_cond)

        if condition is not None:
            assert isinstance(component, pgast.SelectStmt)
            component.where_clause = astutils.extend_binop(
                component.where_clause, condition)

    astutils.for_each_query_in_set(right_rvar.subquery, _inject_filter)
    if not query.from_clause:
        query.from_clause.append(right_rvar)
    else:
        larg = query.from_clause[0]
        rarg = right_rvar

        query.from_clause[0] = pgast.JoinExpr(
            type='cross', larg=larg, rarg=rarg)

def range_for_material_objtype(
    typeref: irast.TypeRef,
    path_id: irast.PathId,
    *,
    for_mutation: bool=False,
    lateral: bool=False,
    include_overlays: bool=True,
    include_descendants: bool=True,
    ignore_rewrites: bool=False,
    dml_source: Optional[irast.MutatingStmt]=None,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:

    env = ctx.env

    if typeref.material_type is not None:
        typeref = typeref.material_type

    relation: Union[pgast.Relation, pgast.CommonTableExpr]

    key = (typeref.id, include_descendants)
    if (
        not ignore_rewrites
        and (rewrite := ctx.env.type_rewrites.get(key)) is not None
        and key not in ctx.pending_type_ctes
        and not for_mutation
    ):

        if (type_cte := ctx.type_ctes.get(key)) is None:
            with ctx.newrel() as sctx:
                sctx.pending_type_ctes.add(key)
                sctx.pending_query = sctx.rel
                sctx.volatility_ref = ()
                sctx.type_rel_overlays = collections.defaultdict(
                    lambda: collections.defaultdict(list))
                sctx.ptr_rel_overlays = collections.defaultdict(
                    lambda: collections.defaultdict(list))
                dispatch.visit(rewrite, ctx=sctx)
                type_cte = pgast.CommonTableExpr(
                    name=ctx.env.aliases.get('t'),
                    query=sctx.rel,
                    materialized=False,
                )
                ctx.type_ctes[key] = type_cte

        with ctx.subrel() as sctx:
            cte_rvar = pgast.RelRangeVar(
                relation=type_cte,
                typeref=typeref,
                alias=pgast.Alias(aliasname=env.aliases.get('t'))
            )
            pathctx.put_path_id_map(sctx.rel, path_id, rewrite.path_id)
            include_rvar(
                sctx.rel, cte_rvar, rewrite.path_id, pull_namespace=False,
                ctx=sctx,
            )
            rvar = rvar_for_rel(
                sctx.rel, lateral=lateral, typeref=typeref, ctx=sctx)
    else:
        assert isinstance(typeref.name_hint, sn.QualName)

        table_schema_name, table_name = common.get_objtype_backend_name(
            typeref.id,
            typeref.name_hint.module,
            aspect=(
                'table' if for_mutation or not include_descendants else
                'inhview'
            ),
            catenate=False,
        )

        if typeref.name_hint.module in {'cfg', 'sys'}:
            # Redirect all queries to schema tables to edgedbss
            table_schema_name = 'edgedbss'

        relation = pgast.Relation(
            schemaname=table_schema_name,
            name=table_name,
            path_id=path_id,
        )

        rvar = pgast.RelRangeVar(
            relation=relation,
            typeref=typeref,
            alias=pgast.Alias(
                aliasname=env.aliases.get(typeref.name_hint.name)
            )
        )

    overlays = get_type_rel_overlays(typeref, dml_source=dml_source, ctx=ctx)
    external_rvar = ctx.env.external_rvars.get((path_id, 'source'))
    if external_rvar is not None:
        if overlays:
            raise AssertionError('cannot mix external and internal overlays')
        return external_rvar

    if overlays and include_overlays:
        set_ops = []

        qry = pgast.SelectStmt()
        qry.from_clause.append(rvar)
        pathctx.put_path_value_rvar(qry, path_id, rvar, env=env)
        if path_id.is_objtype_path():
            pathctx.put_path_source_rvar(qry, path_id, rvar, env=env)
        pathctx.put_path_bond(qry, path_id)

        set_ops.append(('union', qry))

        for op, cte, cte_path_id in overlays:
            rvar = rvar_for_rel(cte, typeref=typeref, ctx=ctx)

            qry = pgast.SelectStmt(
                from_clause=[rvar],
            )

            pathctx.put_path_value_rvar(qry, cte_path_id, rvar, env=env)
            if path_id.is_objtype_path():
                pathctx.put_path_source_rvar(qry, cte_path_id, rvar, env=env)
            pathctx.put_path_bond(qry, cte_path_id)
            pathctx.put_path_id_map(qry, path_id, cte_path_id)

            qry_rvar = pgast.RangeSubselect(
                subquery=qry,
                alias=pgast.Alias(
                    aliasname=env.aliases.get(hint=cte.name or '')
                )
            )

            qry2 = pgast.SelectStmt(
                from_clause=[qry_rvar]
            )
            pathctx.put_path_value_rvar(qry2, path_id, qry_rvar, env=env)
            if path_id.is_objtype_path():
                pathctx.put_path_source_rvar(qry2, path_id, qry_rvar, env=env)
            pathctx.put_path_bond(qry2, path_id)

            if op == 'replace':
                op = 'union'
                set_ops = []
            set_ops.append((op, qry2))

        rvar = range_from_queryset(
            set_ops,
            typeref.name_hint,
            lateral=lateral,
            path_id=path_id,
            tag='overlay-stack',
            ctx=ctx,
        )

    return rvar

def range_for_typeref(
    typeref: irast.TypeRef,
    path_id: irast.PathId,
    *,
    lateral: bool=False,
    for_mutation: bool=False,
    include_descendants: bool=True,
    ignore_rewrites: bool=False,
    dml_source: Optional[irast.MutatingStmt]=None,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:

    if typeref.union:
        # Union object types are represented as a UNION of selects
        # from their children, which is, for most purposes, equivalent
        # to SELECTing from a parent table.
        set_ops = []

        # Concrete unions might have view type elements with duplicate
        # material types, and we need to filter those out.
        seen = set()
        for child in typeref.union:
            mat_child = child.material_type or child
            if mat_child.id in seen:
                assert typeref.union_is_concrete
                continue
            seen.add(mat_child.id)

            c_rvar = range_for_typeref(
                child,
                path_id=path_id,
                include_descendants=not typeref.union_is_concrete,
                for_mutation=for_mutation,
                dml_source=dml_source,
                ctx=ctx,
            )

            qry = pgast.SelectStmt(
                from_clause=[c_rvar],
            )

            pathctx.put_path_value_rvar(qry, path_id, c_rvar, env=ctx.env)
            if path_id.is_objtype_path():
                pathctx.put_path_source_rvar(qry, path_id, c_rvar, env=ctx.env)

            pathctx.put_path_bond(qry, path_id)

            set_ops.append(('union', qry))

        rvar = range_from_queryset(
            set_ops,
            typeref.name_hint,
            lateral=lateral,
            typeref=typeref,
            ctx=ctx,
        )

    elif typeref.intersection:
        wrapper = pgast.SelectStmt()
        component_rvars = []
        for component in typeref.intersection:
            component_rvar = range_for_typeref(
                component,
                lateral=True,
                path_id=path_id,
                for_mutation=for_mutation,
                dml_source=dml_source,
                ctx=ctx,
            )
            pathctx.put_rvar_path_bond(component_rvar, path_id)
            component_rvars.append(component_rvar)
            include_rvar(wrapper, component_rvar, path_id, ctx=ctx)

        int_rvar = pgast.IntersectionRangeVar(component_rvars=component_rvars)
        for aspect in ('source', 'value'):
            pathctx.put_path_rvar(
                wrapper, path_id, int_rvar, aspect=aspect, env=ctx.env
            )

        pathctx.put_path_bond(wrapper, path_id)
        rvar = rvar_for_rel(wrapper, lateral=lateral, ctx=ctx)

    else:
        rvar = range_for_material_objtype(
            typeref,
            path_id,
            lateral=lateral,
            include_descendants=include_descendants,
            ignore_rewrites=ignore_rewrites,
            for_mutation=for_mutation,
            dml_source=dml_source,
            ctx=ctx,
        )

    rvar.query.path_id = path_id

    return rvar

def wrap_set_op_query(
    qry: pgast.SelectStmt, *,
    ctx: context.CompilerContextLevel
) -> pgast.SelectStmt:
    if astutils.is_set_op_query(qry):
        rvar = rvar_for_rel(qry, ctx=ctx)
        nqry = pgast.SelectStmt(from_clause=[rvar])
        nqry.target_list = [
            pgast.ResTarget(
                name=col.name,
                val=pgast.ColumnRef(
                    name=[rvar.alias.aliasname, col.name],
                )
            )
            for col in astutils.get_leftmost_query(qry).target_list
            if col.name
        ]

        pull_path_namespace(target=nqry, source=rvar, ctx=ctx)
        qry = nqry
    return qry

def anti_join(
    lhs: pgast.SelectStmt, rhs: pgast.SelectStmt,
    path_id: Optional[irast.PathId], *,
    ctx: context.CompilerContextLevel,
) -> None:
    """Filter elements out of the LHS that appear on the RHS"""

    if path_id:
        # grab the identity from the LHS and do an
        # anti-join against the RHS.
        src_ref = pathctx.get_path_identity_var(
            lhs, path_id=path_id, env=ctx.env)
        pathctx.get_path_identity_output(
            rhs, path_id=path_id, env=ctx.env)
        cond_expr: pgast.BaseExpr = astutils.new_binop(
            src_ref, rhs, 'NOT IN')
    else:
        # No path we care about. Just check existance.
        cond_expr = pgast.SubLink(
            type=pgast.SubLinkType.NOT_EXISTS, expr=rhs)
    lhs.where_clause = astutils.extend_binop(
        lhs.where_clause, cond_expr)

def range_from_queryset(
    set_ops: Sequence[Tuple[str, pgast.SelectStmt]],
    objname: sn.Name,
    *,
    prep_filter: Callable[
        [pgast.SelectStmt, pgast.SelectStmt], None]=lambda a, b: None,
    path_id: Optional[irast.PathId]=None,
    lateral: bool=False,
    typeref: Optional[irast.TypeRef]=None,
    tag: Optional[str]=None,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:

    rvar: pgast.PathRangeVar

    if len(set_ops) > 1:
        # More than one class table, generate a UNION/EXCEPT clause.
        qry = set_ops[0][1]

        for op, rarg in set_ops[1:]:
            if op == 'filter':
                qry = wrap_set_op_query(qry, ctx=ctx)
                prep_filter(qry, rarg)
                anti_join(qry, rarg, path_id, ctx=ctx)
            else:
                qry = pgast.SelectStmt(
                    op=op,
                    all=True,
                    larg=qry,
                    rarg=rarg,
                )

        rvar = pgast.RangeSubselect(
            subquery=qry,
            lateral=lateral,
            tag=tag,
            alias=pgast.Alias(
                aliasname=ctx.env.aliases.get(objname.name),
            ),
            typeref=typeref,
        )

    else:
        # Just one class table, so return it directly
        from_rvar = set_ops[0][1].from_clause[0]
        assert isinstance(from_rvar, pgast.PathRangeVar)
        rvar = from_rvar

    return rvar

def table_from_ptrref(
    ptrref: irast.PointerRef,
    ptr_info: pg_types.PointerStorageInfo,
    *,
    include_descendants: bool = True,
    for_mutation: bool = False,
    ctx: context.CompilerContextLevel,
) -> pgast.RelRangeVar:
    """Return a Table corresponding to a given Link."""

    aspect = 'table' if for_mutation or not include_descendants else 'inhview'
    table_schema_name, table_name = common.update_aspect(
        ptr_info.table_name, aspect
    )

    if ptrref.name.module in {'cfg', 'sys'}:
        # Redirect all queries to schema tables to edgedbss
        table_schema_name = 'edgedbss'

    relation = pgast.Relation(
        schemaname=table_schema_name, name=table_name)

    # Pseudo pointers (tuple and type intersection) have no schema id.
    sobj_id = ptrref.id if isinstance(ptrref, irast.PointerRef) else None
    typeref = ptrref.out_source if ptrref else None
    rvar = pgast.RelRangeVar(
        schema_object_id=sobj_id,
        typeref=typeref,
        relation=relation,
        alias=pgast.Alias(
            aliasname=ctx.env.aliases.get(ptrref.shortname.name)
        )
    )

    return rvar

def range_for_ptrref(
    ptrref: irast.BasePointerRef, *,
    dml_source: Optional[irast.MutatingStmt]=None,
    for_mutation: bool=False,
    only_self: bool=False,
    path_id: Optional[irast.PathId]=None,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:
    """"Return a Range subclass corresponding to a given ptr step.

    The return value may potentially be a UNION of all tables
    corresponding to a set of specialized links computed from the given
    `ptrref` taking source inheritance into account.
    """

    output_cols = ('source', 'target')

    set_ops = []

    if ptrref.union_components:
        refs = ptrref.union_components
        if only_self and len(refs) > 1:
            raise errors.InternalServerError(
                'unexpected union link'
            )
    elif ptrref.intersection_components:
        # This is a little funky, but in an intersection, the pointer
        # needs to appear in *all* of the tables, so we just pick any
        # one of them.
        refs = {next(iter((ptrref.intersection_components)))}
    elif ptrref.computed_backlink:
        refs = {ptrref.computed_backlink}
    else:
        refs = {ptrref}
        assert isinstance(ptrref, irast.PointerRef), \
            "expected regular PointerRef"
        overlays = get_ptr_rel_overlays(
            ptrref, dml_source=dml_source, ctx=ctx)

    for src_ptrref in refs:
        assert isinstance(src_ptrref, irast.PointerRef), \
            "expected regular PointerRef"

        # Most references to inline links are dispatched to a separate
        # code path (_new_inline_pointer_rvar) by new_pointer_rvar,
        # but when we have union pointers, some might be inline.  We
        # always use the link table if it exists (because this range
        # needs to contain any link properties, for one reason.)
        ptr_info = pg_types.get_ptrref_storage_info(
            src_ptrref, resolve_type=False, link_bias=True,
        )
        if not ptr_info:
            assert ptrref.union_components
            ptr_info = pg_types.get_ptrref_storage_info(
                src_ptrref, resolve_type=False, link_bias=False,
            )

        cols = [
            'source' if ptr_info.table_type == 'link' else 'id',
            ptr_info.column_name,
        ]

        table = table_from_ptrref(
            src_ptrref,
            ptr_info,
            include_descendants=not ptrref.union_is_concrete,
            for_mutation=for_mutation,
            ctx=ctx,
        )

        qry = pgast.SelectStmt()
        qry.from_clause.append(table)

        # Make sure all property references are pulled up properly
        for colname, output_colname in zip(cols, output_cols):
            selexpr = pgast.ColumnRef(
                name=[table.alias.aliasname, colname])
            qry.target_list.append(
                pgast.ResTarget(val=selexpr, name=output_colname))

        set_ops.append(('union', qry))

        overlays = get_ptr_rel_overlays(
            src_ptrref, dml_source=dml_source, ctx=ctx)
        if overlays and not for_mutation:
            # We need the identity var for semi_join to work and
            # the source rvar so that linkprops can be found here.
            if path_id:
                target_ref = qry.target_list[1].val
                pathctx.put_path_identity_var(
                    qry, path_id, var=target_ref, env=ctx.env)
                pathctx.put_path_source_rvar(
                    qry, path_id, table, env=ctx.env)

            for op, cte, cte_path_id in overlays:
                rvar = rvar_for_rel(cte, ctx=ctx)

                qry = pgast.SelectStmt(
                    target_list=[
                        pgast.ResTarget(
                            val=pgast.ColumnRef(
                                name=[col]
                            )
                        )
                        for col in cols
                    ],
                    from_clause=[rvar],
                )
                # Set up identity var, source rvar for reasons discussed above
                if path_id:
                    target_ref = pgast.ColumnRef(
                        name=[rvar.alias.aliasname, cols[1]])
                    pathctx.put_path_identity_var(
                        qry, cte_path_id, var=target_ref, env=ctx.env)
                    pathctx.put_path_source_rvar(
                        qry, cte_path_id, rvar, env=ctx.env)
                    pathctx.put_path_id_map(qry, path_id, cte_path_id)

                set_ops.append((op, qry))

    def prep_filter(larg: pgast.SelectStmt, rarg: pgast.SelectStmt) -> None:
        # Set up the proper join on the source field and clear the target list
        # of the rhs of a filter overlay.
        assert isinstance(larg.target_list[0].val, pgast.ColumnRef)
        assert isinstance(rarg.target_list[0].val, pgast.ColumnRef)
        rarg.where_clause = astutils.join_condition(
            larg.target_list[0].val, rarg.target_list[0].val)
        rarg.target_list.clear()

    return range_from_queryset(
        set_ops, ptrref.shortname,
        prep_filter=prep_filter, path_id=path_id, ctx=ctx)

def range_for_pointer(
    pointer: irast.Pointer,
    *,
    dml_source: Optional[irast.MutatingStmt] = None,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:

    path_id = pointer.target.path_id.ptr_path()
    external_rvar = ctx.env.external_rvars.get((path_id, 'source'))
    if external_rvar is not None:
        return external_rvar

    ptrref = pointer.ptrref
    if ptrref.material_ptr is not None:
        ptrref = ptrref.material_ptr

    return range_for_ptrref(
        ptrref, dml_source=dml_source, path_id=path_id, ctx=ctx)

def rvar_for_rel(
    rel: Union[pgast.BaseRelation, pgast.CommonTableExpr],
    *,
    typeref: Optional[irast.TypeRef] = None,
    lateral: bool = False,
    colnames: Optional[List[str]] = None,
    ctx: context.CompilerContextLevel,
) -> pgast.PathRangeVar:

    rvar: pgast.PathRangeVar

    if colnames is None:
        colnames = []

    if isinstance(rel, pgast.Query):
        alias = ctx.env.aliases.get(rel.name or 'q')

        rvar = pgast.RangeSubselect(
            subquery=rel,
            alias=pgast.Alias(aliasname=alias, colnames=colnames),
            lateral=lateral,
            typeref=typeref,
        )
    else:
        alias = ctx.env.aliases.get(rel.name or '')

        rvar = pgast.RelRangeVar(
            relation=rel,
            alias=pgast.Alias(aliasname=alias, colnames=colnames),
            typeref=typeref,
        )

    return rvar

def _add_type_rel_overlay(
        typeid: uuid.UUID,
        op: str,
        rel: Union[pgast.BaseRelation, pgast.CommonTableExpr], *,
        dml_stmts: Iterable[irast.MutatingStmt] = (),
        path_id: irast.PathId,
        ctx: context.CompilerContextLevel) -> None:
    entry = (op, rel, path_id)
    if dml_stmts:
        for dml_stmt in dml_stmts:
            overlays = ctx.type_rel_overlays[dml_stmt][typeid]
            if entry not in overlays:
                overlays.append(entry)
    else:
        overlays = ctx.type_rel_overlays[None][typeid]
        if entry not in overlays:
            overlays.append(entry)

def add_type_rel_overlay(
        typeref: irast.TypeRef,
        op: str,
        rel: Union[pgast.BaseRelation, pgast.CommonTableExpr], *,
        stop_ref: Optional[irast.TypeRef]=None,
        dml_stmts: Iterable[irast.MutatingStmt] = (),
        path_id: irast.PathId,
        ctx: context.CompilerContextLevel) -> None:
    typeref = typeref.real_material_type
    objs = [typeref]
    if typeref.ancestors:
        objs.extend(typeref.ancestors)

    for obj in objs:
        if stop_ref and (
            obj == stop_ref or
            (stop_ref.ancestors and obj in stop_ref.ancestors)
        ):
            continue
        _add_type_rel_overlay(
            obj.id, op, rel,
            dml_stmts=dml_stmts, path_id=path_id, ctx=ctx)

def get_type_rel_overlays(
    typeref: irast.TypeRef,
    *,
    dml_source: Optional[irast.MutatingStmt]=None,
    ctx: context.CompilerContextLevel,
) -> List[
    Tuple[
        str,
        Union[pgast.BaseRelation, pgast.CommonTableExpr],
        irast.PathId,
    ]
]:
    if typeref.material_type is not None:
        typeref = typeref.material_type

    return ctx.type_rel_overlays[dml_source][typeref.id]

def reuse_type_rel_overlays(
    *,
    dml_stmts: Iterable[irast.MutatingStmt] = (),
    dml_source: irast.MutatingStmt,
    ctx: context.CompilerContextLevel,
) -> None:
    """Update type rel overlays when a DML statement is reused.

    When a WITH bound DML is used, we need to add it (and all of its
    nested overlays) as an overlay for all the enclosing DML
    statements.
    """
    ref_overlays = ctx.type_rel_overlays[dml_source]
    for tid, overlays in ref_overlays.items():
        for op, rel, path_id in overlays:
            _add_type_rel_overlay(
                tid, op, rel, dml_stmts=dml_stmts, path_id=path_id, ctx=ctx
            )
    ptr_overlays = ctx.ptr_rel_overlays[dml_source]
    for (obj, ptr), poverlays in ptr_overlays.items():
        for op, rel, path_id in poverlays:
            _add_ptr_rel_overlay(
                obj, ptr, op, rel, path_id=path_id, dml_stmts=dml_stmts,
                ctx=ctx
            )

def _add_ptr_rel_overlay(
        typeid: uuid.UUID,
        ptrref_name: str,
        op: str,
        rel: Union[pgast.BaseRelation, pgast.CommonTableExpr], *,
        dml_stmts: Iterable[irast.MutatingStmt] = (),
        path_id: irast.PathId,
        ctx: context.CompilerContextLevel) -> None:

    entry = (op, rel, path_id)
    if dml_stmts:
        for dml_stmt in dml_stmts:
            overlays = ctx.ptr_rel_overlays[dml_stmt][typeid, ptrref_name]
            if entry not in overlays:
                overlays.append(entry)
    else:
        overlays = ctx.ptr_rel_overlays[None][typeid, ptrref_name]
        if entry not in overlays:
            overlays.append(entry)

def add_ptr_rel_overlay(
        ptrref: irast.PointerRef,
        op: str,
        rel: Union[pgast.BaseRelation, pgast.CommonTableExpr], *,
        dml_stmts: Iterable[irast.MutatingStmt] = (),
        path_id: irast.PathId,
        ctx: context.CompilerContextLevel) -> None:

    typeref = ptrref.out_source.real_material_type
    objs = [typeref]
    if typeref.ancestors:
        objs.extend(typeref.ancestors)

    for obj in objs:
        _add_ptr_rel_overlay(
            obj.id, ptrref.shortname.name, op, rel, path_id=path_id,
            dml_stmts=dml_stmts,
            ctx=ctx)

def get_ptr_rel_overlays(
    ptrref: irast.PointerRef, *,
    dml_source: Optional[irast.MutatingStmt]=None,
    ctx: context.CompilerContextLevel,
) -> List[
    Tuple[
        str,
        Union[pgast.BaseRelation, pgast.CommonTableExpr],
        irast.PathId,
    ]
]:
    typeref = ptrref.out_source.real_material_type
    return ctx.ptr_rel_overlays[dml_source][typeref.id, ptrref.shortname.name]

def clone_ptr_rel_overlays(
    *,
    ctx: context.CompilerContextLevel,
) -> None:
    ctx.ptr_rel_overlays = ctx.ptr_rel_overlays.copy()
    for k, v in ctx.ptr_rel_overlays.items():
        ctx.ptr_rel_overlays[k] = v.copy()
        for k2, v2 in v.items():
            v[k2] = list(v2)


text:
"""Media helper functions and classes for Google Cloud Storage JSON API."""

from __future__ import absolute_import

import copy
import cStringIO
import httplib
import logging
import socket
import types
import urlparse

import httplib2
from httplib2 import parse_uri

from gslib.cloud_api import BadRequestException
from gslib.progress_callback import ProgressCallbackWithBackoff
from gslib.third_party.storage_apitools import exceptions as apitools_exceptions
from gslib.util import SSL_TIMEOUT
from gslib.util import TRANSFER_BUFFER_SIZE

class BytesTransferredContainer(object):
  """Container class for passing number of bytes transferred to lower layers.

  For resumed transfers or connection rebuilds in the middle of a transfer, we
  need to rebuild the connection class with how much we've transferred so far.
  For uploads, we don't know the total number of bytes uploaded until we've
  queried the server, but we need to create the connection class to pass to
  httplib2 before we can query the server. This container object allows us to
  pass a reference into Upload/DownloadCallbackConnection.
  """

  def __init__(self):
    self.__bytes_transferred = 0

  @property
  def bytes_transferred(self):
    return self.__bytes_transferred

  @bytes_transferred.setter
  def bytes_transferred(self, value):
    self.__bytes_transferred = value

class UploadCallbackConnectionClassFactory(object):
  """Creates a class that can override an httplib2 connection.

  This is used to provide progress callbacks and disable dumping the upload
  payload during debug statements. It can later be used to provide on-the-fly
  hash digestion during upload.
  """

  def __init__(self, bytes_uploaded_container,
               buffer_size=TRANSFER_BUFFER_SIZE,
               total_size=0, progress_callback=None):
    self.bytes_uploaded_container = bytes_uploaded_container
    self.buffer_size = buffer_size
    self.total_size = total_size
    self.progress_callback = progress_callback

  def GetConnectionClass(self):
    """Returns a connection class that overrides send."""
    outer_bytes_uploaded_container = self.bytes_uploaded_container
    outer_buffer_size = self.buffer_size
    outer_total_size = self.total_size
    outer_progress_callback = self.progress_callback

    class UploadCallbackConnection(httplib2.HTTPSConnectionWithTimeout):
      """Connection class override for uploads."""
      bytes_uploaded_container = outer_bytes_uploaded_container
      # After we instantiate this class, apitools will check with the server
      # to find out how many bytes remain for a resumable upload.  This allows
      # us to update our progress once based on that number.
      processed_initial_bytes = False
      GCS_JSON_BUFFER_SIZE = outer_buffer_size
      callback_processor = None
      size = outer_total_size

      def __init__(self, *args, **kwargs):
        kwargs['timeout'] = SSL_TIMEOUT
        httplib2.HTTPSConnectionWithTimeout.__init__(self, *args, **kwargs)

      def send(self, data):
        """Overrides HTTPConnection.send."""
        if not self.processed_initial_bytes:
          self.processed_initial_bytes = True
          if outer_progress_callback:
            self.callback_processor = ProgressCallbackWithBackoff(
                outer_total_size, outer_progress_callback)
            self.callback_processor.Progress(
                self.bytes_uploaded_container.bytes_transferred)
        # httplib.HTTPConnection.send accepts either a string or a file-like
        # object (anything that implements read()).
        if isinstance(data, basestring):
          full_buffer = cStringIO.StringIO(data)
        else:
          full_buffer = data
        partial_buffer = full_buffer.read(self.GCS_JSON_BUFFER_SIZE)
        while partial_buffer:
          httplib2.HTTPSConnectionWithTimeout.send(self, partial_buffer)
          send_length = len(partial_buffer)
          if self.callback_processor:
            # This is the only place where gsutil has control over making a
            # callback, but here we can't differentiate the metadata bytes
            # (such as headers and OAuth2 refreshes) sent during an upload
            # from the actual upload bytes, so we will actually report
            # slightly more bytes than desired to the callback handler.
            #
            # One considered/rejected alternative is to move the callbacks
            # into the HashingFileUploadWrapper which only processes reads on
            # the bytes. This has the disadvantages of being removed from
            # where we actually send the bytes and unnecessarily
            # multi-purposing that class.
            self.callback_processor.Progress(send_length)
          partial_buffer = full_buffer.read(self.GCS_JSON_BUFFER_SIZE)

    return UploadCallbackConnection

def WrapUploadHttpRequest(upload_http):
  """Wraps upload_http so we only use our custom connection_type on PUTs.

  POSTs are used to refresh oauth tokens, and we don't want to process the
  data sent in those requests.

  Args:
    upload_http: httplib2.Http instance to wrap
  """
  request_orig = upload_http.request
  def NewRequest(uri, method='GET', body=None, headers=None,
                 redirections=httplib2.DEFAULT_MAX_REDIRECTS,
                 connection_type=None):
    if method == 'PUT' or method == 'POST':
      override_connection_type = connection_type
    else:
      override_connection_type = None
    return request_orig(uri, method=method, body=body,
                        headers=headers, redirections=redirections,
                        connection_type=override_connection_type)
  # Replace the request method with our own closure.
  upload_http.request = NewRequest

class DownloadCallbackConnectionClassFactory(object):
  """Creates a class that can override an httplib2 connection.

  This is used to provide progress callbacks, disable dumping the download
  payload during debug statements, and provide on-the-fly hash digestion during
  download. On-the-fly digestion is particularly important because httplib2
  will decompress gzipped content on-the-fly, thus this class provides our
  only opportunity to calculate the correct hash for an object that has a
  gzip hash in the cloud.
  """

  def __init__(self, bytes_downloaded_container,
               buffer_size=TRANSFER_BUFFER_SIZE, total_size=0,
               progress_callback=None, digesters=None):
    self.buffer_size = buffer_size
    self.total_size = total_size
    self.progress_callback = progress_callback
    self.digesters = digesters
    self.bytes_downloaded_container = bytes_downloaded_container

  def GetConnectionClass(self):
    """Returns a connection class that overrides getresponse."""

    class DownloadCallbackConnection(httplib2.HTTPSConnectionWithTimeout):
      """Connection class override for downloads."""
      outer_total_size = self.total_size
      outer_digesters = self.digesters
      outer_progress_callback = self.progress_callback
      outer_bytes_downloaded_container = self.bytes_downloaded_container
      processed_initial_bytes = False
      callback_processor = None

      def __init__(self, *args, **kwargs):
        kwargs['timeout'] = SSL_TIMEOUT
        httplib2.HTTPSConnectionWithTimeout.__init__(self, *args, **kwargs)

      def getresponse(self, buffering=False):
        """Wraps an HTTPResponse to perform callbacks and hashing.

        In this function, self is a DownloadCallbackConnection.

        Args:
          buffering: Unused. This function uses a local buffer.

        Returns:
          HTTPResponse object with wrapped read function.
        """
        orig_response = httplib.HTTPConnection.getresponse(self)
        if orig_response.status not in (httplib.OK, httplib.PARTIAL_CONTENT):
          return orig_response
        orig_read_func = orig_response.read

        def read(amt=None):  # pylint: disable=invalid-name
          """Overrides HTTPConnection.getresponse.read.

          This function only supports reads of TRANSFER_BUFFER_SIZE or smaller.

          Args:
            amt: Integer n where 0 < n <= TRANSFER_BUFFER_SIZE. This is a
                 keyword argument to match the read function it overrides,
                 but it is required.

          Returns:
            Data read from HTTPConnection.
          """
          if not amt or amt > TRANSFER_BUFFER_SIZE:
            raise BadRequestException(
                'Invalid HTTP read size %s during download, expected %s.' %
                (amt, TRANSFER_BUFFER_SIZE))
          else:
            amt = amt or TRANSFER_BUFFER_SIZE

          if not self.processed_initial_bytes:
            self.processed_initial_bytes = True
            if self.outer_progress_callback:
              self.callback_processor = ProgressCallbackWithBackoff(
                  self.outer_total_size, self.outer_progress_callback)
              self.callback_processor.Progress(
                  self.outer_bytes_downloaded_container.bytes_transferred)

          data = orig_read_func(amt)
          read_length = len(data)
          if self.callback_processor:
            self.callback_processor.Progress(read_length)
          if self.outer_digesters:
            for alg in self.outer_digesters:
              self.outer_digesters[alg].update(data)
          return data
        orig_response.read = read

        return orig_response
    return DownloadCallbackConnection

def WrapDownloadHttpRequest(download_http):
  """Overrides download request functions for an httplib2.Http object.

  Args:
    download_http: httplib2.Http.object to wrap / override.

  Returns:
    Wrapped / overridden httplib2.Http object.
  """

  # httplib2 has a bug https://code.google.com/p/httplib2/issues/detail?id=305
  # where custom connection_type is not respected after redirects.  This
  # function is copied from httplib2 and overrides the request function so that
  # the connection_type is properly passed through.
  # pylint: disable=protected-access,g-inconsistent-quotes,unused-variable
  # pylint: disable=g-equals-none,g-doc-return-or-yield
  # pylint: disable=g-short-docstring-punctuation,g-doc-args
  # pylint: disable=too-many-statements
  def OverrideRequest(self, conn, host, absolute_uri, request_uri, method,
                      body, headers, redirections, cachekey):
    """Do the actual request using the connection object.

    Also follow one level of redirects if necessary.
    """

    auths = ([(auth.depth(request_uri), auth) for auth in self.authorizations
              if auth.inscope(host, request_uri)])
    auth = auths and sorted(auths)[0][1] or None
    if auth:
      auth.request(method, request_uri, headers, body)

    (response, content) = self._conn_request(conn, request_uri, method, body,
                                             headers)

    if auth:
      if auth.response(response, body):
        auth.request(method, request_uri, headers, body)
        (response, content) = self._conn_request(conn, request_uri, method,
                                                 body, headers)
        response._stale_digest = 1

    if response.status == 401:
      for authorization in self._auth_from_challenge(
          host, request_uri, headers, response, content):
        authorization.request(method, request_uri, headers, body)
        (response, content) = self._conn_request(conn, request_uri, method,
                                                 body, headers)
        if response.status != 401:
          self.authorizations.append(authorization)
          authorization.response(response, body)
          break

    if (self.follow_all_redirects or (method in ["GET", "HEAD"])
        or response.status == 303):
      if self.follow_redirects and response.status in [300, 301, 302,
                                                       303, 307]:
        # Pick out the location header and basically start from the beginning
        # remembering first to strip the ETag header and decrement our 'depth'
        if redirections:
          if not response.has_key('location') and response.status != 300:
            raise httplib2.RedirectMissingLocation(
                "Redirected but the response is missing a Location: header.",
                response, content)
          # Fix-up relative redirects (which violate an RFC 2616 MUST)
          if response.has_key('location'):
            location = response['location']
            (scheme, authority, path, query, fragment) = parse_uri(location)
            if authority == None:
              response['location'] = urlparse.urljoin(absolute_uri, location)
          if response.status == 301 and method in ["GET", "HEAD"]:
            response['-x-permanent-redirect-url'] = response['location']
            if not response.has_key('content-location'):
              response['content-location'] = absolute_uri
            httplib2._updateCache(headers, response, content, self.cache,
                                  cachekey)
          if headers.has_key('if-none-match'):
            del headers['if-none-match']
          if headers.has_key('if-modified-since'):
            del headers['if-modified-since']
          if ('authorization' in headers and
              not self.forward_authorization_headers):
            del headers['authorization']
          if response.has_key('location'):
            location = response['location']
            old_response = copy.deepcopy(response)
            if not old_response.has_key('content-location'):
              old_response['content-location'] = absolute_uri
            redirect_method = method
            if response.status in [302, 303]:
              redirect_method = "GET"
              body = None
            (response, content) = self.request(
                location, redirect_method, body=body, headers=headers,
                redirections=redirections-1,
                connection_type=conn.__class__)
            response.previous = old_response
        else:
          raise httplib2.RedirectLimit(
              "Redirected more times than redirection_limit allows.",
              response, content)
      elif response.status in [200, 203] and method in ["GET", "HEAD"]:
        # Don't cache 206's since we aren't going to handle byte range
        # requests
        if not response.has_key('content-location'):
          response['content-location'] = absolute_uri
        httplib2._updateCache(headers, response, content, self.cache,
                              cachekey)

    return (response, content)

  # Wrap download_http so we do not use our custom connection_type
  # on POSTS, which are used to refresh oauth tokens. We don't want to
  # process the data received in those requests.
  request_orig = download_http.request
  def NewRequest(uri, method='GET', body=None, headers=None,
                 redirections=httplib2.DEFAULT_MAX_REDIRECTS,
                 connection_type=None):
    if method == 'POST':
      return request_orig(uri, method=method, body=body,
                          headers=headers, redirections=redirections,
                          connection_type=None)
    else:
      return request_orig(uri, method=method, body=body,
                          headers=headers, redirections=redirections,
                          connection_type=connection_type)

  # Replace the request methods with our own closures.
  download_http._request = types.MethodType(OverrideRequest, download_http)
  download_http.request = NewRequest

  return download_http

class HttpWithDownloadStream(httplib2.Http):
  """httplib2.Http variant that only pushes bytes through a stream.

  httplib2 handles media by storing entire chunks of responses in memory, which
  is undesirable particularly when multiple instances are used during
  multi-threaded/multi-process copy. This class copies and then overrides some
  httplib2 functions to use a streaming copy approach that uses small memory
  buffers.
  """

  def __init__(self, stream=None, *args, **kwds):
    if stream is None:
      raise apitools_exceptions.InvalidUserInputError(
          'Cannot create HttpWithDownloadStream with no stream')
    self._stream = stream
    self._logger = logging.getLogger()
    super(HttpWithDownloadStream, self).__init__(*args, **kwds)

  @property
  def stream(self):
    return self._stream

  # pylint: disable=too-many-statements
  def _conn_request(self, conn, request_uri, method, body, headers):
    i = 0
    seen_bad_status_line = False
    while i < httplib2.RETRIES:
      i += 1
      try:
        if hasattr(conn, 'sock') and conn.sock is None:
          conn.connect()
        conn.request(method, request_uri, body, headers)
      except socket.timeout:
        raise
      except socket.gaierror:
        conn.close()
        raise httplib2.ServerNotFoundError(
            'Unable to find the server at %s' % conn.host)
      except httplib2.ssl_SSLError:
        conn.close()
        raise
      except socket.error, e:
        err = 0
        if hasattr(e, 'args'):
          err = getattr(e, 'args')[0]
        else:
          err = e.errno
        if err == httplib2.errno.ECONNREFUSED:  # Connection refused
          raise
      except httplib.HTTPException:
        # Just because the server closed the connection doesn't apparently mean
        # that the server didn't send a response.
        if hasattr(conn, 'sock') and conn.sock is None:
          if i < httplib2.RETRIES-1:
            conn.close()
            conn.connect()
            continue
          else:
            conn.close()
            raise
        if i < httplib2.RETRIES-1:
          conn.close()
          conn.connect()
          continue
      try:
        response = conn.getresponse()
      except httplib.BadStatusLine:
        # If we get a BadStatusLine on the first try then that means
        # the connection just went stale, so retry regardless of the
        # number of RETRIES set.
        if not seen_bad_status_line and i == 1:
          i = 0
          seen_bad_status_line = True
          conn.close()
          conn.connect()
          continue
        else:
          conn.close()
          raise
      except (socket.error, httplib.HTTPException):
        if i < httplib2.RETRIES-1:
          conn.close()
          conn.connect()
          continue
        else:
          conn.close()
          raise
      else:
        content = ''
        if method == 'HEAD':
          conn.close()
          response = httplib2.Response(response)
        else:
          if response.status in (httplib.OK, httplib.PARTIAL_CONTENT):
            content_length = None
            if hasattr(response, 'msg'):
              content_length = response.getheader('content-length')
            http_stream = response
            bytes_read = 0
            while True:
              new_data = http_stream.read(TRANSFER_BUFFER_SIZE)
              if new_data:
                self.stream.write(new_data)
                bytes_read += len(new_data)
              else:
                break

            if (content_length is not None and
                long(bytes_read) != long(content_length)):
              # The input stream terminated before we were able to read the
              # entire contents, possibly due to a network condition. Set
              # content-length to indicate how many bytes we actually read.
              self._logger.log(
                  logging.DEBUG, 'Only got %s bytes out of content-length %s '
                  'for request URI %s. Resetting content-length to match '
                  'bytes read.', bytes_read, content_length, request_uri)
              response.msg['content-length'] = str(bytes_read)
            response = httplib2.Response(response)
          else:
            # We fall back to the current httplib2 behavior if we're
            # not processing bytes (eg it's a redirect).
            content = response.read()
            response = httplib2.Response(response)
            # pylint: disable=protected-access
            content = httplib2._decompressContent(response, content)
      break
    return (response, content)

redpajama_set_name: RedPajamaGithub


text:
<?php
class Year extends CI_Controller {

  public function index() {
    exit ('No direct script access allowed');
  }

  /* List years */
  public function get($type = '') {
    // Load helpers
    $this->load->helper(array('year_helper', 'output_helper'));
    switch ($type) {
      case 'yearly':
        echo getTopAlbumByYear($_REQUEST);
        break;
      default:
        echo getYears($_REQUEST);
        break;
    }

  }

  /* Add a year */
  public function add() {
    // Load helpers
    header('HTTP/1.1 501 Not Implemented');
  }

  /* Update year information */
  public function update($type) {
    switch ($type) {
      case 'biography':
        // Load helpers
        $this->load->helper(array('year_helper', 'lastfm_helper'));

        $_REQUEST += fetchTagBio($_REQUEST);
        echo addYearBio($_REQUEST);
        break;
      default:
        break;
    }
  }

  /* Delete year information */
  public function delete() {
    // Load helpers
    header('HTTP/1.1 501 Not Implemented');
  }
}
?>


text:
<?php
/**
 * Backend Page Controller.
 *

 * @package backend.controllers
 *
 */
class BesliderController extends BeController{

        public function __construct($id,$module=null)
	{
		 parent::__construct($id,$module);

                 $this->menu=array(

				  array('label'=>t('Manage Slider'), 'url'=>array('admin','id'=>$id),'linkOptions'=>array('class'=>'btn btn-mini')),

                );

	}

        /**
	 * The function that do Create new Page
	 *
	 */
	public function actionCreate()
	{

		$this->render('slider_create');
	}

        /**
	 * The function that do Manage Page
	 *
	 */
	public function actionAdmin()
	{

	 $this->menu=array_merge($this->menu,
                        array(
                            array('label'=>t('Add photos into Slider'), 'url'=>array('create'),'linkOptions'=>array('class'=>'btn btn-mini')),

                        )
                    );
		$this->render('slider_admin');
	}

        /**
	 * The function that view Page details
	 *
	 */
	public function actionView()
	{
              $id=isset($_GET['id']) ? (int) ($_GET['id']) : 0 ;
              $this->render('slider_view');
	}

        /**
	 * The function that update Page
	 *
	 */
	public function actionUpdate()
	{
                $id=isset($_GET['id']) ? (int) ($_GET['id']) : 0 ;
				$this->render('slider_update',array('id'=>$id));
	}

	public function actionSort()
	{

		$this->render('slider_sort');
	}

        /**
	 * The function is to Delete Page
	 *
	 */
	public function actionDelete($id)
	{
             GxcHelpers::deleteModel('Slider', $id);
	}

}


text:
graphite-golang
===============

This is a lightweight Graphite API client written in Go that implements Carbon
submission functionality.

## Installation

Use `go-get` to install graphite-golang
```
go get github.com/marpaia/graphite-golang
```

## External dependencies

This project has no external dependencies other than the Go standard library.

## Documentation

Like most every other Golang project, this projects documentation can be found
on godoc at [godoc.org/github.com/marpaia/graphite-golang](http://godoc.org/github.com/marpaia/graphite-golang).

## Examples

```go
package mylib

import (
    "github.com/marpaia/graphite-golang"
    "log"
)

func init() {

    // load your configuration file / mechanism
    config := newConfig()

    // try to connect a graphite server
    if config.GraphiteEnabled {
        Graphite, err = graphite.NewGraphite(config.Graphite.Host, config.Graphite.Port)
    } else {
        Graphite = graphite.NewGraphiteNop(config.Graphite.Host, config.Graphite.Port)
    }
    // if you couldn't connect to graphite, use a nop
    if err != nil {
        Graphite = graphite.NewGraphiteNop(config.Graphite.Host, config.Graphite.Port)
    }

    log.Printf("Loaded Graphite connection: %#v", Graphite)
    Graphite.SimpleSend("stats.graphite_loaded", "1")
}

func doWork() {
    // this will work just fine, regardless of if you're working with a graphite
    // nop or not
    Graphite.SimpleSend("stats.doing_work", "1")
}
```

## Contributing

Please contribute and help improve this project!

- Fork the repo
- Make sure the tests pass
- Improve the code
- Make sure your feature has test coverage
- Make sure the tests pass
- Submit a pull request


text:
require "rubygems"
require "bundler"
Bundler.setup

require "config_accessor"


text:
class Die
  def initialize(sides)

   if sides == []
      raise ArgumentError.new('Dude you need to put something in!')
    end
    @sides = sides
  end

  def sides
    @sides.length
  end

  def roll
    @sides.sample
  end
end

# Refactored Solution

class Die
  def initialize(sides)

    raise ArgumentError.new('Dude you need to put something in!') if sides == []
    @sides = sides
  end

  def sides
    @sides.length
  end

  def roll
    @sides.sample
  end
end

# Reflection
=begin
What were the main differences between this die class and the last one you created in terms of implementation?

There argument error condition was differnt, in this challenge we changed it to "if the input is an empty array".  And since we are inputing an array, we needed to change how ruby reads how many sides by using .length. Also since we are inputting an array we had to use .sample method.

Did you need to change much logic to get this to work?

Not much, I knew the input was different and that the built in methods that were used only worked for integers so I had to change them or convert them

What does this exercise teach you about making code that is easily changeable or modifiable?

That keeping everything readible and dry, makes changes easy.

What new methods did you learn when working on this challenge, if any?

The sample method is similar to the random method, but allows us to work with arrays.

What concepts about classes were you able to solidify in this challenge?

That instance variables can be used within the class and more practice with arguement errors was nice,

=end


text:
What's new in h5py 2.6
======================

Support for HDF5 Virtual Dataset API
------------------------------------
Initial support for the HDF5 Virtual Dataset API, which was introduced in
HDF5 1.10, was added to the low-level API. Ideas and input for how this should
work as part of the high-level interface are welcome.

This work was added in :pr:`663` by Aleksandar Jelenak.

Add MPI Collective I/O Support
------------------------------
Support for using MPI Collective I/O in both low-level and high-level code has
been added. See the collective_io.py example for a simple demonstration of how
to use MPI Collective I/O with the high level API.

This work was added in :pr:`648` by Jialin Liu.

Numerous build/testing/CI improvements
--------------------------------------
There were a number of improvements to the setup.py file, which should mean that
`pip install h5py` should work in most places. Work was also done to clean up
the current testing system, using tox is the recommended way of testing h5py
across different Python versions. See :pr:`576` by Jakob Lombacher, :pr:`640` by
Lawrence Mitchell, and :pr:`650`, :pr:`651` and :pr:`658` by James Tocknell.

Cleanup of codebase based on pylint
-----------------------------------
There was a large cleanup of pylint-identified problems by Andrew Collette
(:pr:`578`, :pr:`579`).

Fixes to low-level API
----------------------
Fixes to the typing of functions were added in :pr:`597` by Ulrik Kofoed
Pedersen, :pr:`589` by Peter Chang, and :pr:`625` by Spaghetti Sort. A fix for
variable-length arrays was added in :pr:`621` by Sam Mason. Fixes to compound
types were added in :pr:`639` by @nevion and :pr:`606` by Yu Feng. Finally, a fix
to type conversion was added in :pr:`614` by Andrew Collette.

Documentation improvements
--------------------------
* Updates to FAQ by Dan Guest (:pr:`608`) and Peter Hill (:pr:`607`).
* Updates MPI-related documentation by Jens Timmerman (:pr:`604`) and
  Matthias König (:pr:`572`).
* Fixes to documentation building by Ghislain Antony Vaillant (:pr:`562`,
  :pr:`561`).
* Update PyTables link (:pr:`574` by Dominik Kriegner)
* Add File opening modes to docstring (:pr:`563` by Antony Lee)

Other changes
-------------
* Add `Dataset.ndim` (:pr:`649`, :pr:`660` by @jakirkham, :pr:`661` by James Tocknell)
* Fix import errors in IPython completer (:pr:`605` by Niru Maheswaranathan)
* Turn off error printing in new threads (:pr:`583` by Andrew Collette)
* Use item value in `KeyError` instead of error message (:pr:`642` by Matthias Geier)

Acknowledgements
----------------


text:
package feedbackapi.com.androidfeedbackapitemplate;

import android.renderscript.Double2;
import android.support.annotation.FloatRange;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.view.ViewGroup;
import android.widget.LinearLayout;
import android.widget.RatingBar;
import android.widget.TextView;
import android.widget.Toast;

public class MainActivity extends AppCompatActivity {
    RatingBar ratingBar,ratingBar2;
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        ratingBar = new RatingBar(this);
        ratingBar.setRating(0);
        ratingBar.setStepSize(Float.parseFloat("0.5"));
        ratingBar.setNumStars(5);
        ratingBar.setLayoutParams(new ViewGroup.LayoutParams(ViewGroup.LayoutParams.WRAP_CONTENT, ViewGroup.LayoutParams.WRAP_CONTENT));

        TextView textView = new TextView(this);
        textView.setText("Nutrition");
        textView.setLayoutParams(new ViewGroup.LayoutParams(ViewGroup.LayoutParams.MATCH_PARENT, ViewGroup.LayoutParams.WRAP_CONTENT));

        LinearLayout linearLayout = (LinearLayout)findViewById(R.id.ratingLayout);
        linearLayout.addView(textView);
        linearLayout.addView(ratingBar);

        ratingBar2 = new RatingBar(this);
        ratingBar2.setRating(0);
        ratingBar2.setStepSize(Float.parseFloat("0.5"));
        ratingBar2.setNumStars(5);
        ratingBar2.setLayoutParams(new ViewGroup.LayoutParams(ViewGroup.LayoutParams.WRAP_CONTENT, ViewGroup.LayoutParams.WRAP_CONTENT));

        TextView textView2 = new TextView(this);
        textView2.setText("Healthy Food");
        textView2.setLayoutParams(new ViewGroup.LayoutParams(ViewGroup.LayoutParams.MATCH_PARENT, ViewGroup.LayoutParams.WRAP_CONTENT));

        linearLayout.addView(textView2);
        linearLayout.addView(ratingBar2);

        ratingBar.setOnRatingBarChangeListener(new RatingBar.OnRatingBarChangeListener() {
            @Override
            public void onRatingChanged(RatingBar ratingBar, float rating, boolean fromUser) {
                calcualteRating();
            }
        });

        ratingBar2.setOnRatingBarChangeListener(new RatingBar.OnRatingBarChangeListener() {
            @Override
            public void onRatingChanged(RatingBar ratingBar, float rating, boolean fromUser) {
                calcualteRating();
            }
        });

    }

    private void calcualteRating() {
        final TextView overall = (TextView)findViewById(R.id.overallRatingText);
        double totalRating = ratingBar.getRating() + ratingBar2.getRating();
        overall.setText(totalRating/2.0 + "");
    }

}