Function execute_with_aggregate

src/cypher/cypher.c:3629–3675

Execute WITH aggregation path */

/* Execute WITH aggregation path */
static void execute_with_aggregate(cbm_return_clause_t *wc, binding_t *bindings, int bind_count,
                                   binding_t **vbindings, int *vcount) {
    int agg_cap = CBM_SZ_256;
    with_agg_t *aggs = calloc(agg_cap, sizeof(with_agg_t));
    int agg_cnt = 0;

    for (int bi = 0; bi < bind_count; bi++) {
        char key[CBM_SZ_1K] = "";
        with_agg_build_key(wc, &bindings[bi], key, sizeof(key));
        int found = with_agg_find_or_create(&aggs, &agg_cnt, &agg_cap, wc, &bindings[bi], key);
        with_agg_accumulate(&aggs[found], wc, &bindings[bi]);
    }

    *vbindings = safe_realloc(*vbindings, (agg_cnt + SKIP_ONE) * sizeof(binding_t));
    if (!*vbindings) {
        with_agg_free(aggs, agg_cnt, wc->count);
        return;
    }
    for (int a = 0; a < agg_cnt; a++) {
        binding_t vb = {0};
        /* Carry the store so node_prop can re-fetch a carried node's properties
         * (and compute in_degree/out_degree) on the projected virtual binding. */
        vb.store = (bind_count > 0) ? bindings[0].store : NULL;
        for (int ci = 0; ci < wc->count; ci++) {
            char name_buf[CBM_SZ_256];
            const char *alias = resolve_item_alias(&wc->items[ci], name_buf, sizeof(name_buf));
            if (wc->items[ci].func) {
                char vbuf[CBM_SZ_64];
                if (wc->items[ci].distinct && strcmp(wc->items[ci].func, "COUNT") == 0) {
                    snprintf(vbuf, sizeof(vbuf), "%d", aggs[a].distinct_n[ci]); /* #239 */
                } else {
                    with_agg_format(wc->items[ci].func, &aggs[a], ci, vbuf, sizeof(vbuf));
                }
                with_add_vbinding_var(&vb, alias, vbuf);
            } else {
                with_add_vbinding_var(&vb, alias, aggs[a].group_vals[ci]);
                /* Tag the carried virtual var with the node id (when the group
                 * var is a node) so node_prop can re-fetch its full properties. */
                if (aggs[a].group_node_ids[ci] > 0 && vb.var_count > 0) {
                    vb.var_nodes[vb.var_count - 1].id = aggs[a].group_node_ids[ci];
                }
            }
        }
        (*vbindings)[(*vcount)++] = vb;
    }
    with_agg_free(aggs, agg_cnt, wc->count);
}

/* Execute WITH simple (non-aggregate) projection */
static void execute_with_simple(cbm_return_clause_t *wc, binding_t *bindings, int bind_count,