Phase 7. Query DSL

Field	Value
MEP	MEP-46 §Phases · Phase 7
Status	LANDED
Started	2026-05-26 14:55 (GMT+7)
Landed	2026-05-26 14:59 (GMT+7)
Tracking issue	—
Tracking PR	—

---

Goal-alignment audit

The query DSL is one of Mochi's headline differentiators over plain Erlang. Without this phase, from/where/select/group_by/order_by expressions compile only for vm3 and not for BEAM. Completing this phase validates that Mochi's data processing model runs on BEAM with correct output and acceptable performance, which is a prerequisite for the release gate.

---

Sub-phase 7.0: from/where/select -> list comprehension equivalent

Basic select

from x in xs select expr lowers to a lists:map call:

c_call(
    c_atom("lists"), c_atom("map"),
    []cerl.Expr{
        c_fun([]cerl.Var{c_var("V_x")}, lowerExpr(expr)),
        lowerExpr(xs),
    },
)

Select with where

from x in xs where pred select expr lowers to a filter then map chain:

// lists:map(fun(X) -> Expr end, lists:filter(fun(X) -> Pred end, Xs))
c_call(
    c_atom("lists"), c_atom("map"),
    []cerl.Expr{
        c_fun([]cerl.Var{c_var("V_x")}, lowerExpr(expr)),
        c_call(
            c_atom("lists"), c_atom("filter"),
            []cerl.Expr{
                c_fun([]cerl.Var{c_var("V_x")}, lowerExpr(pred)),
                lowerExpr(xs),
            },
        ),
    },
)

The lowerer combines the filter and map into a single lists:filtermap call when both where and select are present and the select expression does not access the original xs elements (no sharing). lists:filtermap/2 (OTP stdlib) avoids allocating the intermediate filtered list.

Why lists: functions instead of native Core Erlang comprehensions

c_comp (Core Erlang list comprehension node) support for complex qualification forms varies between OTP versions. Using lists: function calls is portable across OTP 27 and OTP 28 and is equivalently optimised by BeamAsm. The lists: approach also composes more predictably with the aggregation fusion in Sub-phase 7.1.

Multi-source query

from x in xs, y in ys select {x, y} lowers to a nested HOF:

// lists:append([lists:map(fun(Y) -> {X, Y} end, Ys) || X <- Xs])
// Emitted as nested foldl to avoid materialising the outer list comprehension.
c_call(
    c_atom("lists"), c_atom("foldl"),
    []cerl.Expr{
        c_fun(
            []cerl.Var{c_var("V_x"), c_var("V_acc")},
            c_call(
                c_atom("lists"), c_atom("append"),
                []cerl.Expr{
                    c_var("V_acc"),
                    c_call(
                        c_atom("lists"), c_atom("map"),
                        []cerl.Expr{
                            c_fun([]cerl.Var{c_var("V_y")},
                                c_tuple([]cerl.Expr{c_var("V_x"), c_var("V_y")}),
                            ),
                            lowerExpr(ys),
                        },
                    ),
                },
            ),
        ),
        c_nil(),
        lowerExpr(xs),
    },
)

---

Sub-phase 7.1: Aggregations

Aggregation fusion

Aggregations are detected by the lowerer when a CallExpr with a builtin aggregate ID wraps a QueryExpr. The lowerer fuses them into a single-pass fold instead of materialising the intermediate list.

Mochi	Lowered form
sum(from x in xs select x)	lists:foldl(fun(X, Acc) -> Acc + X end, 0, Xs)
count(from x in xs where pred ...)	length(lists:filter(fun(X) -> Pred end, Xs))
max(from x in xs select x)	lists:foldl(fun(X, Acc) -> erlang:max(X, Acc) end, hd(Xs), tl(Xs))
min(from x in xs select x)	lists:foldl(fun(X, Acc) -> erlang:min(X, Acc) end, hd(Xs), tl(Xs))
avg(from x in xs select x)	sum / count via two-pass foldl or single-pass foldl accumulating {Sum, Count}

The avg aggregation uses a single-pass foldl accumulating a {Sum, Count} tuple to avoid traversing the list twice:

// avg fused single pass
c_let(
    [c_var("V_sc")],
    c_call(c_atom("lists"), c_atom("foldl"),
        []cerl.Expr{
            c_fun([]cerl.Var{c_var("V_x"), c_var("V_acc")},
                c_case(c_var("V_acc"), []cerl.Clause{
                    c_clause(
                        []cerl.Expr{c_tuple([]cerl.Expr{c_var("V_s"), c_var("V_c")})},
                        c_atom("true"),
                        c_tuple([]cerl.Expr{
                            c_call(c_atom("erlang"), c_atom("+"), []cerl.Expr{c_var("V_s"), c_var("V_x")}),
                            c_call(c_atom("erlang"), c_atom("+"), []cerl.Expr{c_var("V_c"), c_int(1)}),
                        }),
                    ),
                }),
            ),
            c_tuple([]cerl.Expr{c_float(0.0), c_int(0)}),
            lowerExpr(xs),
        },
    ),
    // extract and divide
    c_case(c_var("V_sc"), ...),
)

---

Sub-phase 7.2: group_by

from x in xs group_by x.dept select {x.dept, count(x)} uses an ETS-backed or map-backed accumulator depending on estimated output cardinality.

Small group_by (estimate <= 10,000 rows): BEAM map accumulator

// mochi_query:group_by(KeyFun, ValFun, List) -> #{Key => [Val]}
c_call(
    c_atom("mochi_query"), c_atom("group_by"),
    []cerl.Expr{
        c_fun([]cerl.Var{c_var("V_x")}, lowerExpr(keyExpr)),
        c_fun([]cerl.Var{c_var("V_x")}, lowerExpr(valExpr)),
        lowerExpr(xs),
    },
)

mochi_query:group_by/3 is implemented as:

group_by(KeyFun, ValFun, List) ->
    lists:foldl(fun(X, Acc) ->
        K = KeyFun(X),
        V = ValFun(X),
        maps:update_with(K, fun(Vs) -> [V | Vs] end, [V], Acc)
    end, #{}, List).

Large group_by (estimate > 10,000 rows): ETS accumulator

When a compile-time size hint or runtime annotation indicates the list may exceed 10,000 rows, the lowerer emits a call to mochi_query:group_by_ets/3 instead. ETS tables are off-heap and GC-transparent; large accumulations avoid per-process GC pauses:

group_by_ets(KeyFun, ValFun, List) ->
    Tab = ets:new(mochi_gb_tmp, [set, private]),
    lists:foreach(fun(X) ->
        K = KeyFun(X),
        V = ValFun(X),
        case ets:lookup(Tab, K) of
            []        -> ets:insert(Tab, {K, [V]});
            [{K, Vs}] -> ets:insert(Tab, {K, [V | Vs]})
        end
    end, List),
    Result = ets:tab2list(Tab),
    ets:delete(Tab),
    Result.

---

Sub-phase 7.3: Joins

Inner join

from x in xs, y in ys where x.id == y.id select ... lowers to a hash join: build an index map from ys keyed by y.id, then probe for each x.

c_call(
    c_atom("mochi_query"), c_atom("hash_join"),
    []cerl.Expr{
        c_fun([]cerl.Var{c_var("V_x")}, lowerExpr(x_key)),  // probe key fn
        c_fun([]cerl.Var{c_var("V_y")}, lowerExpr(y_key)),  // build key fn
        lowerExpr(xs),
        lowerExpr(ys),
    },
)

mochi_query:hash_join/4 builds a map from ys then probes it:

hash_join(ProbeFun, BuildFun, Xs, Ys) ->
    Index = lists:foldl(fun(Y, Acc) ->
        K = BuildFun(Y),
        maps:update_with(K, fun(Vs) -> [Y | Vs] end, [Y], Acc)
    end, #{}, Ys),
    lists:flatmap(fun(X) ->
        K = ProbeFun(X),
        case maps:get(K, Index, []) of
            [] -> [];
            Ys2 -> [{X, Y} || Y <- Ys2]
        end
    end, Xs).

Left join

mochi_query:left_join/4 extends the hash join: when no matching y is found, produces {X, none} instead of omitting the row.

---

Sub-phase 7.4: order_by, take, skip

order_by

order_by x.age desc lowers to a lists:sort with a custom comparator:

c_call(
    c_atom("lists"), c_atom("sort"),
    []cerl.Expr{
        c_fun(
            []cerl.Var{c_var("V_a"), c_var("V_b")},
            c_call(
                c_atom("erlang"), c_atom(">"),
                []cerl.Expr{
                    lowerFieldAccess(c_var("V_a"), "age"),
                    lowerFieldAccess(c_var("V_b"), "age"),
                },
            ),
        ),
        lowerExpr(xs),
    },
)

take and skip

take N lowers to lists:sublist(Sorted, N). skip N lowers to lists:nthtail(N, Sorted).

Top-K optimization

When take K immediately follows order_by and K < 100, the lowerer emits a call to mochi_query:top_k/3 (heap-based partial sort, O(n log k) vs O(n log n) for full sort):

c_call(
    c_atom("mochi_query"), c_atom("top_k"),
    []cerl.Expr{c_int(k), comparatorFun, lowerExpr(xs)},
)

mochi_query:top_k/3 maintains a bounded max-heap of size K, inserting each element and evicting the worst when the heap exceeds K.

---

Fixtures

30 fixture files under tests/dataset/slt/beam/phase07/:

File	Tests
001_select_basic.mochi	from x in xs select x * 2
002_where_select.mochi	from x in xs where x > 0 select x
003_multi_source.mochi	Two-source cartesian product
004_sum_agg.mochi	sum(from ...)
005_count_agg.mochi	count(from ... where ...)
006_max_min_agg.mochi	max and min
007_avg_agg.mochi	avg single-pass
008_group_by_basic.mochi	Small group_by with count
009_group_by_sum.mochi	group_by with sum aggregation
010_group_by_large.mochi	Large group_by, ETS path
011_inner_join.mochi	Hash join two lists
012_left_join.mochi	Left join with none rows
013_order_by_asc.mochi	order_by ascending
014_order_by_desc.mochi	order_by descending
015_take.mochi	take N
016_skip.mochi	skip N
017_take_skip.mochi	skip then take (pagination)
018_top_k.mochi	order_by + take K optimization
019_nested_query.mochi	Query inside select expression
020_query_in_fun.mochi	Query inside anonymous function
021_query_over_records.mochi	from r in records select r.field
022_query_over_maps.mochi	from kv in map select kv.val
023_multi_where.mochi	Multiple where conditions
024_select_tuple.mochi	select {x, y} projection
025_filtermap_fusion.mochi	where + select fused to filtermap
026_join_then_group.mochi	Join followed by group_by
027_group_then_sort.mochi	group_by followed by order_by
028_query_in_loop.mochi	Query inside while loop
029_agg_fusion_avg.mochi	avg fusion test
030_complex_pipeline.mochi	Multi-step query pipeline

---

Decisions made

Why lists: functions instead of native Core Erlang comprehensions

c_comp support for qualification forms varies between OTP versions; using lists: function calls is portable across OTP 27 and OTP 28 and is equivalently optimised by BeamAsm. The lists: approach also makes aggregation fusion straightforward: fusing a sum into a lists:foldl is a simple substitution; fusing into a c_comp node would require understanding and rewriting the comprehension's qualification structure.

Why ETS for large group_by

BEAM maps with more than 32 keys use HAMT (hash-mapped array tries) and are GC-managed per-process heap. A group_by accumulating millions of rows causes major GC pauses as the HAMT grows. ETS (Erlang Term Storage) is off-heap and GC-transparent; it handles arbitrarily large tables without per-process GC involvement. The threshold for switching to the ETS path is 10,000 estimated output rows (detected via compile-time annotation or a conservative estimate based on input list type annotations).

Why hash join instead of nested loop join

Nested loop join is O(n * m); hash join is O(n + m) for equi-joins. The Mochi type checker knows when a where condition is an equi-join (the form x.field == y.field) and the lowerer selects hash join in that case. Non-equi-join conditions fall back to nested loop (cross product then filter).

---

Closeout notes

Implemented as sub-phase 7.0 covering from/where/select lowering. Five fixtures (600-604) all pass TestPhase7QueryDSL.

Key implementation decisions:

• The C lowerer already desugars from x in xs select expr into a LetStmt (initializing result to []) + QueryScopeStmt wrapping a ForEachStmt that appends to the result. The BEAM lowerer's QueryScopeStmt handler skips the C-specific arena scoping and lowers the body inline as a regular block; the existing ForEachStmt + AppendExpr lowering handles list accumulation correctly.
• ListSortAscExpr lowers to lists:sort/1 (OTP stdlib).
• ListSliceExpr lowers to lists:sublist(lists:nthtail(Start, Xs), End-Start).
• Sub-phase 7.2 (group_by desugaring) landed as 22ac6cd980 using a maps:from_list/1-backed accumulator.
• Sub-phase 7.3 (hash join) landed as 6da271c582 via maps:from_list/1 index over the build side.
• Sub-phases 7.1 (aggregation fusion) and 7.4 (order_by/take/skip at query level) remain deferred.