Skip to content

graflo.hq.auto_join

Auto-JOIN generation for edge resources.

When a Resource's pipeline contains an EdgeActor whose derivation declares match_source / match_target, and the source/target vertex types have known table connectors, this module can auto-generate JoinClauses and IS_NOT_NULL filters on the edge resource's table connector so that the resulting SQL fetches fully resolved rows.

enrich_edge_connector_with_joins(resource, connector, bindings, vertex_config)

Mutate connector in-place, adding JoinClauses + IS_NOT_NULL filters.

The function inspects the Resource's actor pipeline for EdgeActors and, for each edge that declares match_source and match_target, looks up the source / target vertex table connectors and primary keys to construct LEFT JOINs and NOT-NULL guards.

If the connector already has joins, this function is a no-op (the user provided explicit join specs).

Parameters:

Name Type Description Default
resource ResourceRuntime | ResourceConfig

Declarative :class:ResourceConfig or initialized :class:ResourceRuntime whose pipeline is inspected.

 required
connector TableConnector

The table connector to enrich (mutated in-place).

 required
bindings Bindings

The Bindings collection holding all vertex table connectors.

 required
vertex_config VertexConfig

VertexConfig for looking up primary keys.

 required
Source code in graflo/hq/auto_join.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def enrich_edge_connector_with_joins(
    resource: ResourceRuntime | ResourceConfig,
    connector: TableConnector,
    bindings: Bindings,
    vertex_config: VertexConfig,
) -> None:
    """Mutate *connector* in-place, adding JoinClauses + IS_NOT_NULL filters.

    The function inspects the Resource's actor pipeline for EdgeActors and,
    for each edge that declares ``match_source`` **and** ``match_target``,
    looks up the source / target vertex table connectors and primary keys to
    construct LEFT JOINs and NOT-NULL guards.

    If the connector already has joins, this function is a no-op (the user
    provided explicit join specs).

    Args:
        resource: Declarative :class:`ResourceConfig` or initialized
            :class:`ResourceRuntime` whose pipeline is inspected.
        connector: The table connector to enrich (mutated in-place).
        bindings: The Bindings collection holding all vertex table connectors.
        vertex_config: VertexConfig for looking up primary keys.
    """
    if connector.joins:
        return

    edge_actors = _collect_edge_actors(_actor_root_for_joins(resource))
    if not edge_actors:
        return

    new_joins: list[JoinClause] = []
    new_filters: list[FilterExpression] = []

    for ea in edge_actors:
        edge = ea.edge
        der = ea.derivation
        if not der.match_source or not der.match_target:
            continue
        # Dynamic EdgeActors (source_type_field set) resolve types at row-processing
        # time; static vertex type info is not available for JOIN generation.
        if edge is None:
            continue

        source_info = _vertex_table_info(edge.source, bindings, vertex_config)
        target_info = _vertex_table_info(edge.target, bindings, vertex_config)
        if source_info is None or target_info is None:
            logger.debug(
                "Skipping auto-join for edge %s->%s: missing vertex connector",
                edge.source,
                edge.target,
            )
            continue

        src_table, src_schema, src_pk = source_info
        tgt_table, tgt_schema, tgt_pk = target_info

        src_alias = _SOURCE_ALIAS
        tgt_alias = _TARGET_ALIAS

        new_joins.append(
            JoinClause(
                table=src_table,
                schema_name=src_schema,
                alias=src_alias,
                on_self=der.match_source,
                on_other=src_pk,
                join_type="LEFT",
            )
        )
        new_joins.append(
            JoinClause(
                table=tgt_table,
                schema_name=tgt_schema,
                alias=tgt_alias,
                on_self=der.match_target,
                on_other=tgt_pk,
                join_type="LEFT",
            )
        )

        new_filters.append(
            FilterExpression(
                kind="leaf",
                field=f"{src_alias}.{src_pk}",
                cmp_operator=ComparisonOperator.IS_NOT_NULL,
            )
        )
        new_filters.append(
            FilterExpression(
                kind="leaf",
                field=f"{tgt_alias}.{tgt_pk}",
                cmp_operator=ComparisonOperator.IS_NOT_NULL,
            )
        )

    if new_joins:
        connector.joins = new_joins
        connector.filters = list(connector.filters) + new_filters