Function _apply_join_projection

dlt/dataset/_join.py:288–344

Apply join projection contract onto `query`. Preserves the left-side projection and appends only columns from the explicitly joined `target_table` as `{projection_prefix}__{column}` aliases. `allow_existing_target_projection` is used for idempotent re-joins: when a join call contri

(
    query: sge.Select,
    *,
    schema: Schema,
    left_table: str,
    target_table: str,
    target_qualifier: str,
    projection_prefix: str,
    allow_existing_target_projection: bool,
)

def _apply_join_projection(
    query: sge.Select,
    *,
    schema: Schema,
    left_table: str,
    target_table: str,
    target_qualifier: str,
    projection_prefix: str,
    allow_existing_target_projection: bool,
) -> None:
    """Apply join projection contract onto `query`.

    Preserves the left-side projection and appends only columns from the explicitly
    joined `target_table` as `{projection_prefix}__{column}` aliases.

    `allow_existing_target_projection` is used for idempotent re-joins: when a
    join call contributes no new join edges, all target-prefixed columns may already
    exist in the left projection and should be accepted as a no-op instead of raising
    a collision error.
    """
    normalized_left_expressions = _normalize_left_projection(query, left_table)

    existing_projection_column_names = {
        expr.output_name
        for expr in normalized_left_expressions
        if expr.output_name not in {"", "*"}
    }

    target_columns = schema.get_table_columns(target_table)
    target_output_names = {
        f"{projection_prefix}__{column_name}" for column_name in target_columns.keys()
    }
    duplicate_output_names = target_output_names & existing_projection_column_names
    if duplicate_output_names:
        if duplicate_output_names == target_output_names and allow_existing_target_projection:
            # no-op: all target columns are already projected (on duplicate join call for example)
            return
        duplicate_names_list = ", ".join(sorted(duplicate_output_names))
        raise ValueError(
            "Join projection output names conflict with existing columns: "
            f"{duplicate_names_list}. Choose a different `alias` for `join(...)`."
        )

    appended_target_columns: list[sge.Expression] = []
    for column_name in target_columns.keys():
        output_name = f"{projection_prefix}__{column_name}"
        appended_target_columns.append(
            sge.Alias(
                this=sge.Column(
                    table=sge.to_identifier(target_qualifier, quoted=False),
                    this=sge.to_identifier(column_name, quoted=True),
                ),
                alias=sge.to_identifier(output_name, quoted=True),
            )
        )

    query.set("expressions", [*normalized_left_expressions, *appended_target_columns])