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 Resource

The Resource 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 
 31
 32
 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
def enrich_edge_connector_with_joins(
    resource: Resource,
    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: The Resource 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(resource.root)
    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