From a96bed65f182f27bd943c6b1a3e0661230e5bee1 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Fri, 13 Nov 2026 17:55:00 -0500 Subject: [PATCH] DDIR: `flatmap` returns complete rows, without positions `flatmap((k, v) -> l)` now takes `l : List((key, val))`, and each element is a whole output row, key included. It no longer keeps the input key or adds each element's position to the value. Positions came from the June port, so `flatmap` and `collect` could round-trip a list; a lambda that wants them now writes `((0, l len) range, l) zip`. The runtime takes keys and values straight from the list's element storage, and repeats only times and diffs. When every list holds at most one row, it gathers times and diffs by the surviving input rows; when every list holds exactly one, it reuses them as they are. Shape and demand analysis, explain, the language reference and README, every DDP program, the Rust fixtures, the Python plan generators and the plan snapshots follow. Of the 25 written uses, 18 now write positions explicitly, and 7 choose their key in the lambda, removing a following map. Co-Authored-By: GPT-6 Sol (Codex) Claude-Session: https://claude.ai/code/session_01H6wB4H99TLw4tqu4x48pXL --- interactive/README.md | 3 + .../bench/ldbc/authored/plans/bi01.ddp | 2 +- .../bench/ldbc/authored/plans/bi12.ddp | 2 +- .../bench/ldbc/authored/plans/bi19.ddp | 2 +- .../bench/ldbc/authored/plans/ic11.ddp | 2 +- .../bench/ldbc/authored/plans/is03.ddp | 2 +- .../bench/ldbc/authored/plans/is07.ddp | 2 +- interactive/bench/ldbc/bi5.ddp | 2 +- interactive/bench/ldbc/ic6.ddp | 2 +- .../interactive_v2/oneoff/ic11.oneoff.ddp | 2 +- interactive/bench/ldbc/is3.ddp | 2 +- interactive/bench/ldbc/snb/rel.py | 2 +- interactive/bench/ldbc/test_topk.py | 2 +- interactive/examples/aoc2023/day03/part1.ddp | 3 +- interactive/examples/aoc2023/day03/part2.ddp | 3 +- interactive/examples/aoc2023/day15/part1.ddp | 3 +- interactive/examples/aoc2023/day15/part2.ddp | 3 +- interactive/examples/aoc2023/day16/part1.ddp | 3 +- interactive/examples/aoc2023/day16/part2.ddp | 3 +- interactive/examples/aoc2023/day18/part1.ddp | 3 +- interactive/examples/aoc2023/day18/part2.ddp | 3 +- interactive/examples/programs/ast.ddp | 2 +- interactive/examples/programs/tour.ddp | 2 +- interactive/examples/programs/unnest.ddp | 2 +- .../examples/server/programs/arrange_idem.ddp | 2 +- interactive/src/backend/corgi.rs | 45 +++++---- interactive/src/explain/mod.rs | 20 ++-- interactive/src/ir.rs | 3 +- interactive/src/lambda.rs | 2 +- interactive/src/parse/mod.rs | 4 +- interactive/src/parse/pipe.rs | 4 +- interactive/src/passes/projection.rs | 7 +- interactive/src/shapes.rs | 12 ++- interactive/tests/difftest.rs | 10 +- interactive/tests/explain.rs | 19 +++- interactive/tests/flatmap.rs | 92 +++++++++++++++++++ .../plan_pairs/expected/ldbc_ic11_oneoff.ddp | 2 +- .../ldbc_ic11_oneoff.lowered--optimized.diff | 4 +- .../ldbc_ic11_oneoff.parsed--lowered.diff | 4 +- .../expected/ldbc_ic11_oneoff.parsed.ddp | 2 +- .../tests/plan_pairs/expected/ldbc_is03.ddp | 2 +- .../ldbc_is03.lowered--optimized.diff | 4 +- .../expected/ldbc_is03.parsed--lowered.diff | 4 +- .../plan_pairs/expected/ldbc_is03.parsed.ddp | 2 +- .../tests/plan_pairs/expected/ldbc_is07.ddp | 2 +- .../ldbc_is07.lowered--optimized.diff | 4 +- .../expected/ldbc_is07.parsed--lowered.diff | 4 +- .../plan_pairs/expected/ldbc_is07.parsed.ddp | 2 +- .../expected/server_arrange_idem.ddp | 2 +- ...erver_arrange_idem.lowered--optimized.diff | 4 +- .../server_arrange_idem.parsed--lowered.diff | 4 +- .../expected/server_arrange_idem.parsed.ddp | 2 +- interactive/tests/programs/f64_math.ddp | 2 +- interactive/tests/programs/lambda_slots.ddp | 4 +- interactive/tests/programs/registered.ddp | 2 +- 55 files changed, 220 insertions(+), 113 deletions(-) create mode 100644 interactive/tests/flatmap.rs diff --git a/interactive/README.md b/interactive/README.md index a47ee9f07..62a2cbcf1 100644 --- a/interactive/README.md +++ b/interactive/README.md @@ -63,6 +63,9 @@ The pair structure `(key, val)` calls out first how the data are organized, and A currently unfortunate amount of symbolic load is in shuffling around fields of collection elements. The `map` operator applies to each record, and the `->` fragment is an expression converting the input to the output. +`flatmap` gives a `List((key, val))` of complete rows, choosing both keys and values. +A zero-or-one-element list drops or keeps a row; `match` builds only the selected arm. +For positions, explicitly zip a list `l` with its range: `((0, l len) range, l) zip`. The language is a small interpreted columnar language with ADTs, and is at the moment a bit esoteric (the `inject` verb puts data within an enum variant). diff --git a/interactive/bench/ldbc/authored/plans/bi01.ddp b/interactive/bench/ldbc/authored/plans/bi01.ddp index 83a2f6dff..5786319ea 100644 --- a/interactive/bench/ldbc/authored/plans/bi01.ddp +++ b/interactive/bench/ldbc/authored/plans/bi01.ddp @@ -43,7 +43,7 @@ let total = sums | filter(((_, _, _, _, c), _) -> (c, 0) eq) | join_function(((rid, _, _, _, _), n) -> ((rid, ()), (0), n) enlist) | count; let ranked = groups | map(((rid, year, kind, cat), (n, s)) -> (rid, (year, kind, cat, n, s))) | join(total, (rid, (year, kind, cat, n, s), all) -> (rid, ((year neg, kind, cat), (n, s, all)))) - | collect | flatmap((_, entries) -> entries); + | collect | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((neg_year, kind, cat), (n, s, all)))) -> ((rid, rank), (neg_year neg, kind, cat, n, (s to_float, n to_float) div, s, ((n, 100) mul to_float, all to_float) div))); diff --git a/interactive/bench/ldbc/authored/plans/bi12.ddp b/interactive/bench/ldbc/authored/plans/bi12.ddp index 26bcd0710..3e565e61e 100644 --- a/interactive/bench/ldbc/authored/plans/bi12.ddp +++ b/interactive/bench/ldbc/authored/plans/bi12.ddp @@ -29,7 +29,7 @@ let counted = chosen | map(((rid, who, _), ()) -> ((rid, who), ())) | count; let perhead = counted + ((everyone - (everyone | join(counted, (key, z, _) -> (key, z))))); let people = perhead | map(((rid, _), n) -> ((rid, n), ())) | count; let ranked = people | map(((rid, n), p) -> (rid, (p neg, n neg))) - | collect | flatmap((_, entries) -> entries); + | collect | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, (np, nn))) -> ((rid, rank), (nn neg, np neg))); output "answer" = answer | arrange; diff --git a/interactive/bench/ldbc/authored/plans/bi19.ddp b/interactive/bench/ldbc/authored/plans/bi19.ddp index 68577ea26..73a0b1c7e 100644 --- a/interactive/bench/ldbc/authored/plans/bi19.ddp +++ b/interactive/bench/ldbc/authored/plans/bi19.ddp @@ -38,7 +38,7 @@ let best = ends | map((rid, (_, _, cost)) -> (rid, cost)) | min; let ranked = ends | join(best, (rid, (source, node, cost), least) -> (rid, (source, node, cost, least))) | filter((_, (_, _, cost, least)) -> (cost, least) eq) | map((rid, (source, node, cost, _)) -> (rid, ((source, node), cost))) - | collect | flatmap((_, entries) -> entries); + | collect | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((source, node), cost))) -> ((rid, rank), (source, node, cost))); output "answer" = answer | arrange; diff --git a/interactive/bench/ldbc/authored/plans/ic11.ddp b/interactive/bench/ldbc/authored/plans/ic11.ddp index b94131d92..6ab0e136e 100644 --- a/interactive/bench/ldbc/authored/plans/ic11.ddp +++ b/interactive/bench/ldbc/authored/plans/ic11.ddp @@ -27,7 +27,7 @@ let jobs = circle | map(((node, rid, country, year), ()) -> (node, (rid, coun | map((key, (company, _, _)) -> (key, company)); -- Within one person and start year, number companies by name; order puts the last name first. -let named = jobs | collect | flatmap((_, entries) -> entries); +let named = jobs | collect | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); -- Top k: `reduce` sees a key's values in sort order and keeps the first k (copies capped -- at k), emitting (rank, value). Changes below the top k produce no output. let ranked = named | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))) diff --git a/interactive/bench/ldbc/authored/plans/is03.ddp b/interactive/bench/ldbc/authored/plans/is03.ddp index fb7fb67ba..a4fb749f4 100644 --- a/interactive/bench/ldbc/authored/plans/is03.ddp +++ b/interactive/bench/ldbc/authored/plans/is03.ddp @@ -7,7 +7,7 @@ let friend = import "graph::friend" : (int, (int, int)); -- (person, (frie let ranked = request | map(((rid, who), ()) -> (who, rid)) | join(friend, (_, rid, (f, since)) -> (f, (rid, since))) | join(person, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))) - | collect | flatmap((_, entries) -> entries); + | collect | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); output "answer" = answer | arrange; diff --git a/interactive/bench/ldbc/authored/plans/is07.ddp b/interactive/bench/ldbc/authored/plans/is07.ddp index de69d3a90..0e88a18c8 100644 --- a/interactive/bench/ldbc/authored/plans/is07.ddp +++ b/interactive/bench/ldbc/authored/plans/is07.ddp @@ -24,7 +24,7 @@ let known = reply | map(((who, op), row) -> (who, (op, row))) | map(((pair, row), ()) -> (pair, row)); let marked = (known | map(((who, _), row) -> (who, (row, 1)))) + ((reply - known) | map(((who, _), row) -> (who, (row, 0)))); let ranked = marked | join(person, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))) - | collect | flatmap((_, entries) -> entries); + | collect | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); output "answer" = answer | arrange; diff --git a/interactive/bench/ldbc/bi5.ddp b/interactive/bench/ldbc/bi5.ddp index 7a6936291..c79c9462f 100644 --- a/interactive/bench/ldbc/bi5.ddp +++ b/interactive/bench/ldbc/bi5.ddp @@ -18,5 +18,5 @@ let replied = selected | map(((rid, m, creator), ()) -> (m, (rid, creator))) let totals = (own + liked + replied) | collect | map((k, vs) -> (k, ((0, 0, 0), vs) fold (((a0, a1, a2), (x0, x1, x2)) -> ((a0, x0) add, (a1, x1) add, (a2, x2) add)))); let scores = totals | map(((rid, creator), t) -> let (m, r, l) = t in (rid, (((m, (2, r) mul) add, (10, l) mul) add neg, creator, t))); -output "answer" = scores | collect | flatmap((_, vs) -> vs) | filter((_, (pos, _)) -> (pos, 100) lt) +output "answer" = scores | collect | flatmap((k, vs) -> (k, ((0, vs len) range, vs) zip) cap_list) | filter((_, (pos, _)) -> (pos, 100) lt) | map((rid, (pos, (score, creator, (m, r, l)))) -> ((rid, pos), (creator, r, l, m, score neg))); diff --git a/interactive/bench/ldbc/ic6.ddp b/interactive/bench/ldbc/ic6.ddp index 4cbefbbd1..4f1618587 100644 --- a/interactive/bench/ldbc/ic6.ddp +++ b/interactive/bench/ldbc/ic6.ddp @@ -19,5 +19,5 @@ let counts = selected | map(((rid, m, tag), ()) -> ((m), (rid, tag))) | join(tagged, (_, (rid, tag), (name)) -> ((rid, name, tag), ())) | filter(((_, name, tag), ()) -> (name, tag) ne) | map(((rid, name, _), ()) -> ((rid, name), ())) | count; output "answer" = counts | map(((rid, name), n) -> (rid, (n neg, name))) - | collect | flatmap((_, vs) -> vs) | filter((_, (pos, _)) -> (pos, 10) lt) + | collect | flatmap((k, vs) -> (k, ((0, vs len) range, vs) zip) cap_list) | filter((_, (pos, _)) -> (pos, 10) lt) | map((rid, (pos, (negn, name))) -> ((rid, pos), (name, negn neg))); diff --git a/interactive/bench/ldbc/interactive_v2/oneoff/ic11.oneoff.ddp b/interactive/bench/ldbc/interactive_v2/oneoff/ic11.oneoff.ddp index be5f084a7..c2673ec19 100644 --- a/interactive/bench/ldbc/interactive_v2/oneoff/ic11.oneoff.ddp +++ b/interactive/bench/ldbc/interactive_v2/oneoff/ic11.oneoff.ddp @@ -17,7 +17,7 @@ let jobs = circle | map(((node, rid, country, year), ()) -> (node, (rid, coun | join(place, (_, (key, (company, country)), (there, _, _)) -> (key, (company, country, there))) | filter((_, (_, country, there)) -> (country, there) eq) | map((key, (company, _, _)) -> (key, company)); -let named = jobs | collect | flatmap((_, entries) -> entries); +let named = jobs | collect | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let ranked = named | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))) | join(person, (node, (rid, since, pos, company), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since, node, pos neg), (first, last, company)))) | reduce((_, vals) -> let positive = (vals map ((_, d) -> (d, 0) gt), (0, vals len) range) zip filter in let indices = ((0, (positive len, 10) min) range, positive) gather in let prefix = (indices, vals) gather in let ends = prefix map ((_, d) -> (d, 10) min) scan_add in let chosen = (prefix, ends) zip map (((item, d), end) -> let count = (d, 10) min in let before = (end, count) sub in let keep = (((10, before) sub, 0) max, count) min in ((keep, 0) gt, (item, keep))) filter in let ends = chosen map ((_, d) -> d) scan_add in (chosen, ends) zip map (((item, d), end) -> (item, ((end, d) sub, end) range) cap_list map ((item, rank) -> ((rank, item), 1))) flatten .1); diff --git a/interactive/bench/ldbc/is3.ddp b/interactive/bench/ldbc/is3.ddp index f39eebdf6..b5d9685de 100644 --- a/interactive/bench/ldbc/is3.ddp +++ b/interactive/bench/ldbc/is3.ddp @@ -5,5 +5,5 @@ let person = import "graph::person" : ((int), (List(int), List(int), int)); let friends = request | map(((rid, pid), ()) -> ((pid), rid)) | join(edge, (_, rid, (q, created)) -> ((q), (rid, created))) | join(person, ((q), (rid, created), (first, last, _)) -> (rid, (created neg, q, first, last))); -output "answer" = friends | collect | flatmap((_, vs) -> vs) +output "answer" = friends | collect | flatmap((k, vs) -> (k, ((0, vs len) range, vs) zip) cap_list) | map((rid, (pos, (negc, q, first, last))) -> ((rid, pos), (q, first, last, negc neg))); diff --git a/interactive/bench/ldbc/snb/rel.py b/interactive/bench/ldbc/snb/rel.py index e07f205b1..03f4dc5ed 100644 --- a/interactive/bench/ldbc/snb/rel.py +++ b/interactive/bench/ldbc/snb/rel.py @@ -399,7 +399,7 @@ def keyed(source, fields, keys, keep=True): ordered = f'{src} | map((row, ()) -> ({projection([col(n) for n in keys],refs)}, {sort}))' out = product([*(f'item.{len(order)}.{i}' for i in range(len(source.fields))),'rank']) if limit is None: - code = ordered+' | collect | flatmap((_, values) -> values)' + code = ordered+' | collect | flatmap((k, values) -> (k, ((0, values len) range, values) zip) cap_list)' code += f' | map((_, (rank, item)) -> ({out}, ()))' else: # DD retains all candidates; the lambda copies only the best diff --git a/interactive/bench/ldbc/test_topk.py b/interactive/bench/ldbc/test_topk.py index 084a93caf..4f6370828 100644 --- a/interactive/bench/ldbc/test_topk.py +++ b/interactive/bench/ldbc/test_topk.py @@ -40,7 +40,7 @@ def program(): exports.append(f'output "{name}" = {compiler.emit(ranked)} | arrange;') lines = [*compiler.lines, *exports, f'let values = {initial} ' '| map((row, ()) -> (row.0, (row.1, row.2, row)));', - 'let baseline = values | collect | flatmap((_, vals) -> vals) ' + 'let baseline = values | collect | flatmap((k, vals) -> (k, ((0, vals len) range, vals) zip) cap_list) ' '| filter((_, (rank, _)) -> (rank, 20) lt);', 'output "baseline" = baseline | map((key, value) -> ((key, value.0), value.1)) | arrange;'] for name, (buckets, hash_key) in VARIANTS.items(): diff --git a/interactive/examples/aoc2023/day03/part1.ddp b/interactive/examples/aoc2023/day03/part1.ddp index b71c85444..25ca92f0e 100644 --- a/interactive/examples/aoc2023/day03/part1.ddp +++ b/interactive/examples/aoc2023/day03/part1.ddp @@ -29,8 +29,7 @@ labels: { -- digit cells adjacent to a symbol let symnbrs = symbols - | flatmap(_ -> [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 0), (0, 1), (1, -1), (1, 0), (1, 1)]) - | map(((r, col), (_, (dr, dc))) -> (((r, dr) add, (col, dc) add), ())) | distinct; + | flatmap(((r, col), _) -> ((r, col), [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 0), (0, 1), (1, -1), (1, 0), (1, 1)]) cap_list map (((r, col), (dr, dc)) -> (((r, dr) add, (col, dc) add), ()))) | distinct; let active_runs = labels::label | map(((r, col, s), _) -> ((r, col), s)) | join(symnbrs, ((r, _), s, _) -> ((r, s), ())) | distinct; diff --git a/interactive/examples/aoc2023/day03/part2.ddp b/interactive/examples/aoc2023/day03/part2.ddp index a54bf525e..8fb5dd723 100644 --- a/interactive/examples/aoc2023/day03/part2.ddp +++ b/interactive/examples/aoc2023/day03/part2.ddp @@ -27,8 +27,7 @@ labels: { let labelcells = labels::label | map(((r, col, s), _) -> ((r, col), s)); let gearnbrs = gears - | flatmap(_ -> [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 0), (0, 1), (1, -1), (1, 0), (1, 1)]) - | map(((r, col), (_, (dr, dc))) -> (((r, dr) add, (col, dc) add), (r, col))); + | flatmap(((r, col), _) -> ((r, col), [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 0), (0, 1), (1, -1), (1, 0), (1, 1)]) cap_list map (((r, col), (dr, dc)) -> (((r, dr) add, (col, dc) add), (r, col)))); -- (nbr_cell, gear) let pairs = gearnbrs diff --git a/interactive/examples/aoc2023/day15/part1.ddp b/interactive/examples/aoc2023/day15/part1.ddp index a37d50da1..50299487d 100644 --- a/interactive/examples/aoc2023/day15/part1.ddp +++ b/interactive/examples/aoc2023/day15/part1.ddp @@ -7,8 +7,7 @@ let chars = input 0 : ((int, int, int), ()) | map(((r, pos, code), ()) -> ((r, p let lens = chars | map(((r, _), _) -> (r, ())) | count; -- (r, len) let ktab = chars | map(_ -> ((), ())) | distinct - | flatmap(_ -> (0, 34) range) - | map((_, (_, k)) -> ((), k)); + | flatmap(_ -> (0, 34) range map (k -> ((), k))); -- ((), k) hash: { diff --git a/interactive/examples/aoc2023/day15/part2.ddp b/interactive/examples/aoc2023/day15/part2.ddp index c2efd205a..b4a9f4c1c 100644 --- a/interactive/examples/aoc2023/day15/part2.ddp +++ b/interactive/examples/aoc2023/day15/part2.ddp @@ -5,8 +5,7 @@ let ops = input 0 : ((int, int, int, int, int), ()) | map(((r, l1, l2, op, d), ()) -> (r, (l1, l2, ((op, 61) eq, (d, 48) sub, 0) select))); let ktab = ops | map(_ -> ((), ())) | distinct - | flatmap(_ -> (0, 34) range) - | map((_, (_, k)) -> ((), k)); + | flatmap(_ -> (0, 34) range map (k -> ((), k))); let opsk = ops | map((r, (l1, l2, f)) -> ((l1, l2), (r, f))); -- ((l1,l2), (r, op)) let dels = opsk | filter((_, (_, f)) -> (f, 0) eq) | map((l, (r, _)) -> (l, (1000000, r) sub)) | min diff --git a/interactive/examples/aoc2023/day16/part1.ddp b/interactive/examples/aoc2023/day16/part1.ddp index edb8edb71..e8c0dbdee 100644 --- a/interactive/examples/aoc2023/day16/part1.ddp +++ b/interactive/examples/aoc2023/day16/part1.ddp @@ -10,8 +10,7 @@ let shift = cells | map(_ -> ((), ())) | distinct | flatmap(_ -> [ (1, 124, 1, 0, 3), (1, 124, -1, 0, 2), (1, 47, 1, 0, 3), (1, 92, -1, 0, 2), (2, 46, -1, 0, 2), (2, 45, 0, 1, 0), (2, 45, 0, -1, 1), (2, 124, -1, 0, 2), (2, 47, 0, 1, 0), (2, 92, 0, -1, 1), (3, 46, 1, 0, 3), (3, 45, 0, 1, 0), - (3, 45, 0, -1, 1), (3, 124, 1, 0, 3), (3, 47, 0, -1, 1), (3, 92, 0, 1, 0) ]) - | map((_, (_, (d, s, dr, dc, nd))) -> ((d, s), (dr, dc, nd))); + (3, 45, 0, -1, 1), (3, 124, 1, 0, 3), (3, 47, 0, -1, 1), (3, 92, 0, 1, 0) ] map ((d, s, dr, dc, nd) -> ((d, s), (dr, dc, nd)))); -- ((dir, sym), (dr, dc, newdir)) beam: { diff --git a/interactive/examples/aoc2023/day16/part2.ddp b/interactive/examples/aoc2023/day16/part2.ddp index 0698e67d6..6b621e840 100644 --- a/interactive/examples/aoc2023/day16/part2.ddp +++ b/interactive/examples/aoc2023/day16/part2.ddp @@ -10,8 +10,7 @@ let shift = cells | map(_ -> ((), ())) | distinct | flatmap(_ -> [ (1, 124, 1, 0, 3), (1, 124, -1, 0, 2), (1, 47, 1, 0, 3), (1, 92, -1, 0, 2), (2, 46, -1, 0, 2), (2, 45, 0, 1, 0), (2, 45, 0, -1, 1), (2, 124, -1, 0, 2), (2, 47, 0, 1, 0), (2, 92, 0, -1, 1), (3, 46, 1, 0, 3), (3, 45, 0, 1, 0), - (3, 45, 0, -1, 1), (3, 124, 1, 0, 3), (3, 47, 0, -1, 1), (3, 92, 0, 1, 0) ]) - | map((_, (_, (d, s, dr, dc, nd))) -> ((d, s), (dr, dc, nd))); + (3, 45, 0, -1, 1), (3, 124, 1, 0, 3), (3, 47, 0, -1, 1), (3, 92, 0, 1, 0) ] map ((d, s, dr, dc, nd) -> ((d, s), (dr, dc, nd)))); let rows = cells | map(((r, _), _) -> (r, ())) | distinct; let cols = cells | map(((_, c), _) -> (c, ())) | distinct; diff --git a/interactive/examples/aoc2023/day18/part1.ddp b/interactive/examples/aoc2023/day18/part1.ddp index be591ea61..60fffd518 100644 --- a/interactive/examples/aoc2023/day18/part1.ddp +++ b/interactive/examples/aoc2023/day18/part1.ddp @@ -22,8 +22,7 @@ let n = s | join(perim, (_, a, b) -> ((), (a, b) add)); -- ((), |S| + P), even; halve by binary decomposition (no division) let powtab = instr | map(_ -> ((), ())) | distinct - | flatmap(_ -> [549755813888, 274877906944, 137438953472, 68719476736, 34359738368, 17179869184, 8589934592, 4294967296, 2147483648, 1073741824, 536870912, 268435456, 134217728, 67108864, 33554432, 16777216, 8388608, 4194304, 2097152, 1048576, 524288, 262144, 131072, 65536, 32768, 16384, 8192, 4096, 2048, 1024, 512, 256, 128, 64, 32, 16, 8, 4, 2, 1]) - | map((_, (i, p)) -> (i, p)); + | flatmap(_ -> let l = [549755813888, 274877906944, 137438953472, 68719476736, 34359738368, 17179869184, 8589934592, 4294967296, 2147483648, 1073741824, 536870912, 268435456, 134217728, 67108864, 33554432, 16777216, 8388608, 4194304, 2097152, 1048576, 524288, 262144, 131072, 65536, 32768, 16384, 8192, 4096, 2048, 1024, 512, 256, 128, 64, 32, 16, 8, 4, 2, 1] in ((0, l len) range, l) zip); -- (i, 2^(39-i)) halve: { diff --git a/interactive/examples/aoc2023/day18/part2.ddp b/interactive/examples/aoc2023/day18/part2.ddp index 6ec9fd8e3..af2911ea6 100644 --- a/interactive/examples/aoc2023/day18/part2.ddp +++ b/interactive/examples/aoc2023/day18/part2.ddp @@ -28,8 +28,7 @@ let n = s | join(perim, (_, a, b) -> ((), (a, b) add)); -- ((), |S| + P), even; halve by binary decomposition (no division) let powtab = instr | map(_ -> ((), ())) | distinct - | flatmap(_ -> [549755813888, 274877906944, 137438953472, 68719476736, 34359738368, 17179869184, 8589934592, 4294967296, 2147483648, 1073741824, 536870912, 268435456, 134217728, 67108864, 33554432, 16777216, 8388608, 4194304, 2097152, 1048576, 524288, 262144, 131072, 65536, 32768, 16384, 8192, 4096, 2048, 1024, 512, 256, 128, 64, 32, 16, 8, 4, 2, 1]) - | map((_, (i, p)) -> (i, p)); + | flatmap(_ -> let l = [549755813888, 274877906944, 137438953472, 68719476736, 34359738368, 17179869184, 8589934592, 4294967296, 2147483648, 1073741824, 536870912, 268435456, 134217728, 67108864, 33554432, 16777216, 8388608, 4194304, 2097152, 1048576, 524288, 262144, 131072, 65536, 32768, 16384, 8192, 4096, 2048, 1024, 512, 256, 128, 64, 32, 16, 8, 4, 2, 1] in ((0, l len) range, l) zip); -- (i, 2^(39-i)) halve: { diff --git a/interactive/examples/programs/ast.ddp b/interactive/examples/programs/ast.ddp index dc8cd16a2..f004e35a9 100644 --- a/interactive/examples/programs/ast.ddp +++ b/interactive/examples/programs/ast.ddp @@ -21,7 +21,7 @@ let lists = rows ) append)); -- one row per element, carrying its position -let exploded = lists | flatmap((_, l) -> l); +let exploded = lists | flatmap((k, l) -> (k, ((0, l len) range, l) zip) cap_list); -- bucket by (position, magnitude); payload folds a four-element list, wrapped in a variant -- and matched straight back out (the tag is the point, not the payload). diff --git a/interactive/examples/programs/tour.ddp b/interactive/examples/programs/tour.ddp index 3ed20c696..a90bb64ba 100644 --- a/interactive/examples/programs/tour.ddp +++ b/interactive/examples/programs/tour.ddp @@ -27,7 +27,7 @@ let tagged = edges let folded = edges | map((a, b) -> (a, (a enlist, b enlist, 7 enlist) append)) | map((k, l) -> (k, ((0, l) fold ((acc, x) -> (acc, x) add), l len))); -let nested = edges | flatmap((a, b) -> (b enlist, a enlist) append) | collect; +let nested = edges | flatmap(row -> let (a, b) = row in let l = (b enlist, a enlist) append in (row.0, ((0, l len) range, l) zip) cap_list) | collect; -- Reducers. let counted = edges | count; diff --git a/interactive/examples/programs/unnest.ddp b/interactive/examples/programs/unnest.ddp index 8cba18a1a..9e30ceedb 100644 --- a/interactive/examples/programs/unnest.ddp +++ b/interactive/examples/programs/unnest.ddp @@ -5,7 +5,7 @@ let rows = input 0 : ((int, int), ()) | map(((a, b), ()) -> (a, (a enlist, b enlist, (a, b) add enlist) append)); -- rows: (a, [a, b, a+b]) -let exploded = rows | flatmap((_, l) -> l); +let exploded = rows | flatmap((k, l) -> (k, ((0, l len) range, l) zip) cap_list); -- one row per element: (a, (pos, elem)) let regrouped = exploded | collect; diff --git a/interactive/examples/server/programs/arrange_idem.ddp b/interactive/examples/server/programs/arrange_idem.ddp index 20253e3a5..56d6954e8 100644 --- a/interactive/examples/server/programs/arrange_idem.ddp +++ b/interactive/examples/server/programs/arrange_idem.ddp @@ -23,7 +23,7 @@ let elems = node | map((id, _) -> (id, id)); -- id ; id (s -- canonicalized through the current class map, then regrouped into a signature let pc = node | map((id, (op, cs)) -> ((id, op), cs)) - | flatmap((_, cs) -> cs) + | flatmap((k, cs) -> (k, ((0, cs len) range, cs) zip) cap_list) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); -- (child, (id, op, pos)) -- nodes that produce an arrangement (op 1 = Arrange, op 2 = Reduce) diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index c79baebd7..7cea2c8ab 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -183,29 +183,36 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: Level, plans: &[Plan]) -> CC { } LinearOp::FlatMap(_) => { let g = plan.graph(&kshape, &vshape); - // Structural explode: the evaluated list column's FLAT element storage already - // IS the new value column, so the elements never move. Each row's span in the - // bounds gives both the within-row position (DDIR's `$1[0]`) and a repeat map - // carrying key/time/diff across. No per-row eval, no transcode. + // The flat element storage is already the output rows. Repeat only time/diff. let (bounds, elems) = - corgi::eval_graph(g, CValue::Prod(vec![c.keys.clone(), c.vals])).into_list("flatmap list").unwrap(); - let ends: Vec = bounds.to_vec(); - let total = ends.last().copied().unwrap_or(0); - let (mut reps, mut pos) = (Vec::with_capacity(total), Vec::with_capacity(total)); - let mut start = 0usize; - for (row, end) in ends.into_iter().enumerate() { - for p in 0..(end - start) { - reps.push(row); - pos.push(p as i64); + corgi::eval_graph(g, CValue::Prod(vec![c.keys, c.vals])).into_list("flatmap list").unwrap(); + let mut row = elems.into_prod("flatmap row").unwrap(); + let vals = row.pop().unwrap(); + let keys = row.pop().unwrap(); + // Enlist's uniform 1:1 case can retain the original metadata columns. + if bounds.strided() == Some(1) { + c = CorgiContainer { keys, vals, times: c.times, diffs: c.diffs }; + continue; + } + let ends = bounds.to_vec(); + let mut start = 0; + let zero_or_one = ends.iter().all(|&end| { + let len = end - start; + start = end; + len <= 1 + }); + let mut reps = Vec::with_capacity(ends.last().copied().unwrap_or(0)); + start = 0; + for (r, end) in ends.into_iter().enumerate() { + if zero_or_one { + if end != start { reps.push(r); } + } else { + reps.extend(std::iter::repeat_n(r, end - start)); } start = end; } - CorgiContainer { - keys: gather(&c.keys, &reps), - vals: CValue::Prod(vec![CValue::i64(pos), elems]), - times: c.times.gather(&reps), - diffs: reps.iter().map(|&r| c.diffs[r]).collect(), - } + CorgiContainer { keys, vals, times: c.times.gather(&reps), + diffs: reps.iter().map(|&r| c.diffs[r]).collect() } } }; } diff --git a/interactive/src/explain/mod.rs b/interactive/src/explain/mod.rs index 353203755..1e3d72821 100644 --- a/interactive/src/explain/mod.rs +++ b/interactive/src/explain/mod.rs @@ -493,13 +493,14 @@ impl Sb { fn emit_lookup_flatmap(&mut self, dep_y: Ref, side: &Side, out_user_len: usize, list: &Lambda) -> Ref { let (in_len, out_len) = (side.user_len, out_user_len); let src = self.concat(vec![side.witness.clone(), side.forward.clone()]); - // Pair table: each source row keyed by (k, hash of its value), exploded, then re-keyed - // to the output (k, (pos, elem)) carrying (k, hash). + // Pair each complete output row with (source key, hash of source value). + // Carrying the hash rather than the list avoids quadratic output size. let hashed = || tup(vec![var(0), format!("{} hash", var(1))]); let rekeyed = self.project(src.clone(), projection(hashed(), f(1, 0), 2)); - let exploded = self.linear(rekeyed, vec![LinearOp::FlatMap(list.clone())]); - let pair = self.project(exploded, projection(tup(vec![f(0, 0), var(1)]), var(0), 2)); - // Re-key the demand by the same output (k, (pos, elem)); keep its chain ++ q. + let paired = slot(2, &[(list, tup(vec![var(0), var(1)]))], |out| + format!("({}, {}) cap_list map ((source, row) -> (row, source))", var(0), out[0])); + let pair = self.linear(rekeyed, vec![LinearOp::FlatMap(paired)]); + // Re-key the demand by the complete output; keep its chain ++ q. let dep_keyed = self.project(dep_y, projection(tup(vec![var(0), f(1, 0)]), tup(fidx(1, 1, out_len + 2)), 2)); // The demanded elements' sources, as (k, hash) ; chain_out ++ q. let sourced = self.join(dep_keyed, pair, projection(tup(vec![f(2, 0), f(2, 1)]), tup(fidx(1, 0, out_len + 1)), 3)); @@ -524,11 +525,12 @@ impl Sb { let in_len = side.user_len; let src = self.concat(vec![side.witness.clone(), side.forward.clone()]); let rekeyed = self.project(src, projection(tup(vec![var(0), var(1)]), tup(Vec::new()), 2)); - let list = slot(2, &[(lambda, tup(vec![f(0, 0), format!("{}.1.0", var(0))]))], |out| out[0].clone()); + let list = slot(2, &[(lambda, tup(vec![f(0, 0), format!("{}.1.0", var(0))]))], |out| + format!("({}, {}) cap_list map ((source, row) -> (source, row))", var(0), out[0])); let exploded = self.linear(rekeyed, vec![LinearOp::FlatMap(list)]); - // FlatMap's value is (position, ((key, val), time, diff)). - let nonzero = self.filter(exploded, predicate(format!("({}.1.2, 0) ne", var(1)))); - let pair = self.project(nonzero, projection(format!("{}.1.0", var(1)), tup(vec![var(0), format!("{}.1.1", var(1))]), 2)); + // FlatMap's value is ((key, val), time, diff). + let nonzero = self.filter(exploded, predicate(format!("({}.2, 0) ne", var(1)))); + let pair = self.project(nonzero, projection(format!("{}.0", var(1)), tup(vec![var(0), format!("{}.1", var(1))]), 2)); let dep_keyed = self.project(dep_y, projection(tup(vec![var(0), f(1, 0)]), tup(fidx(1, 1, out_len + 2)), 2)); // Wide value: (V_in, chain_in..., chain_out..., q, time). let source = format!("{}.0", var(2)); diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index 122b1ae57..871f36be7 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -39,8 +39,7 @@ pub enum LinearOp { /// the value. Time itself is unchanged. See `Expr::LiftIter` for the /// discipline restriction. LiftIter, - /// UNNEST: explode the `List` a lambda gives into one row per element, value - /// `(pos, element)`. See `parse::Expr::FlatMap`. + /// Each row becomes the lambda's list of complete `(key, val)` rows. FlatMap(Lambda), /// Pair each value with its whole time: `(key, val)` becomes `(key, (val, time))`. /// The time is a tuple of `Int`s, the epoch and then every scope coordinate, so its diff --git a/interactive/src/lambda.rs b/interactive/src/lambda.rs index 4e1aaca35..e51c4f82d 100644 --- a/interactive/src/lambda.rs +++ b/interactive/src/lambda.rs @@ -395,7 +395,7 @@ fn slot_graph(lambda: &Lambda, shapes: &[Shape]) -> Res<(corgi::Compiled, Shape) /// not list-shaped is the type error: the backend explodes the column structurally. pub(crate) fn compile_flatmap(list: &Lambda, kshape: &Shape, vshape: &Shape) -> Res> { match slot_graph(list, &[kshape.clone(), vshape.clone()])? { - (k, Shape::List(_)) => Ok(k.graph().clone()), + (k, Shape::List(e)) => { crate::shapes::row_shape(*e)?; Ok(k.graph().clone()) }, (_, other) => Err(format!("flatmap over a non-list: {other}")), } } diff --git a/interactive/src/parse/mod.rs b/interactive/src/parse/mod.rs index 06d957a74..e1ed9b063 100644 --- a/interactive/src/parse/mod.rs +++ b/interactive/src/parse/mod.rs @@ -68,9 +68,7 @@ pub enum Expr { /// implicit leave. This preserves the "loop body is a time-invariant /// function" property; in-scope use risks defeating the fixpoint. LiftIter(Box), - /// UNNEST: explode the `List` a lambda gives into one row per element, keyed - /// as the input, with value `(pos, element)` (position innermost-first - /// so `Collect` can restore order). The cross-join of a row with its list. + /// Replace each row with the lambda's list of complete `(key, val)` rows. FlatMap(Box, Lambda), /// Pair each row's value with its whole time. See `LinearOp::Lift`. Lift(Box), diff --git a/interactive/src/parse/pipe.rs b/interactive/src/parse/pipe.rs index fc7540180..cd2cb8f4c 100644 --- a/interactive/src/parse/pipe.rs +++ b/interactive/src/parse/pipe.rs @@ -42,8 +42,8 @@ //! key's first `n` values (`n` at least one), whatever their diffs; `min` is //! `reduce(limit 1, (_, vals) -> (vals head.0, 1) enlist)`. //! - `| filter((k, v) -> mask)` — keep rows where the lambda gives a nonzero `Int`. -//! - `| flatmap((k, v) -> list)` — UNNEST: explode the `List` the lambda gives into one row per -//! element, value `(pos, element)`. +//! - `| flatmap((k, v) -> list)` — any number of complete `(key, val)` rows. +//! For positions, write `((0, l len) range, l) zip`; keys are chosen by the lambda. //! - `| negate` · `| arrange` · `| inspect(label)` · //! `| lift_iter`. //! - `| lift` — pair each value with its whole time, `(key, (val, time))`, the time a tuple diff --git a/interactive/src/passes/projection.rs b/interactive/src/passes/projection.rs index 807122bc5..eede31c4b 100644 --- a/interactive/src/passes/projection.rs +++ b/interactive/src/passes/projection.rs @@ -505,9 +505,8 @@ fn build_scope( LinearOp::LiftIter => builder.lift_iter(&after, &before), LinearOp::Lift => builder.lift(&after, &before), LinearOp::FlatMap(lambda) => { - builder.key_identity(&after, &before); - // The list determines output presence and positions, - // even when no element payload is demanded. + builder.presence(&after, &before); + // The lambda determines output presence as well as rows. builder.scalar_lambda(&after, &[&before], lambda); } LinearOp::JoinFunction(lambda) => { @@ -1144,7 +1143,7 @@ mod tests { fn flatmap_reads_only_the_list_it_explodes() { let text = "let x = input 0 : (int, (List(int), List(int))) | map((k, v) -> (k, v)); - output \"out\" = x | flatmap((k, v) -> v.0) | count;"; + output \"out\" = x | flatmap((k, v) -> (k, v.0) cap_list) | count;"; let mut program = lower::lower_tree(parse::pipe::parse(text)); assert_eq!(pushdown(&mut program, &|_| None).unwrap(), 1); } diff --git a/interactive/src/shapes.rs b/interactive/src/shapes.rs index c54e57f77..dbce9c18a 100644 --- a/interactive/src/shapes.rs +++ b/interactive/src/shapes.rs @@ -250,7 +250,7 @@ pub(crate) fn linear_shapes(ops: &[LinearOp], key: Shape, val: Shape, level: Lev }, LinearOp::Negate => {} LinearOp::FlatMap(t) => match slot_shape(t, &env)? { - Shape::List(e) => v = Shape::Prod(vec![Shape::Int, *e]), + Shape::List(e) => (k, v) = row_shape(*e)?, s => return Err(format!("flatmap of shape {s}, not a list")), }, // Extend a tuple, or pair any other value. @@ -267,14 +267,18 @@ pub(crate) fn linear_shapes(ops: &[LinearOp], key: Shape, val: Shape, level: Lev Ok(steps) } -/// The `(key, val)` shapes a projection gives over rows of `rows`. -fn pair_shape(projection: &Lambda, rows: &[Shape]) -> Result<(Shape, Shape), String> { - match slot_shape(projection, rows)? { +pub(crate) fn row_shape(shape: Shape) -> Result<(Shape, Shape), String> { + match shape { Shape::Prod(mut fs) if fs.len() == 2 => { let v = fs.pop().unwrap(); Ok((fs.pop().unwrap(), v)) } other => Err(format!("gives {other}, not (key, val)")), } } +/// The `(key, val)` shapes a projection gives over rows of `rows`. +fn pair_shape(projection: &Lambda, rows: &[Shape]) -> Result<(Shape, Shape), String> { + row_shape(slot_shape(projection, rows)?) +} + /// The output value shape of `reduce(lambda)` over a key of shape `key` with values of shape `val`: /// the lambda reads the key and a `List` of `(val, diff)`, and must give a `List` of `(val, diff)`. fn reduce_shape(lambda: &Lambda, key: &Shape, val: &Shape) -> Result { diff --git a/interactive/tests/difftest.rs b/interactive/tests/difftest.rs index fd81f16ce..61ae27b93 100644 --- a/interactive/tests/difftest.rs +++ b/interactive/tests/difftest.rs @@ -484,7 +484,7 @@ impl Gen { let elem = if self.rng.chance(70) { Sh::Int } else { self.shape(&env, false, false) }; let list = self.expr(&Sh::List(Box::new(elem.clone())), &env, false); val = Sh::Tup(vec![Sh::Int, elem]); - Op::Plain { text: format!("flatmap({input} -> {list})"), same_shape: false } + Op::Plain { text: format!("flatmap(row -> let {input} = row in let l = {list} in (row.0, ((0, l len) range, l) zip) cap_list)"), same_shape: false } } 55..=59 => same_shape("negate".into()), 60..=62 => same_shape("arrange".into()), @@ -502,7 +502,7 @@ impl Gen { // A rank, as the LDBC queries write one: collect, then explode the list, // each element beside its position. val = Sh::Tup(vec![Sh::Int, elem]); - Op::Plain { text: format!("{once}collect | flatmap((_, l) -> l)"), same_shape: false } + Op::Plain { text: format!("{once}collect | flatmap((k, l) -> (k, ((0, l len) range, l) zip) cap_list)"), same_shape: false } } else { Op::Plain { text: format!("{once}collect"), same_shape: false } } @@ -714,7 +714,7 @@ impl Gen { let mut env = Env::new(); let input = format!("({}, {})", self.pattern(&pk, &mut env), self.pattern(&pv, &mut env)); let list = self.bounded_list(&env); - ops.push(Op::Plain { text: format!("flatmap({input} -> {list})"), same_shape: false }); + ops.push(Op::Plain { text: format!("flatmap(row -> let {input} = row in let l = {list} in (row.0, ((0, l len) range, l) zip) cap_list)"), same_shape: false }); pv = Sh::Tup(vec![Sh::Int, Sh::Int]); } let mut env = Env::new(); @@ -741,7 +741,7 @@ impl Gen { let mut e = Env::new(); let input = format!("({}, ())", self.pattern(key, &mut e)); let list = self.bounded_list(&e); - ops.push(Op::Plain { text: format!("flatmap({input} -> {list})"), same_shape: false }); + ops.push(Op::Plain { text: format!("flatmap(row -> let {input} = row in let l = {list} in (row.0, ((0, l len) range, l) zip) cap_list)"), same_shape: false }); let mut exploded = Env::new(); let row = format!("({}, {})", self.pattern(key, &mut exploded), self.pattern(&Sh::pair(), &mut exploded)); ops.push(Op::Plain { text: format!("map({row} -> ({}, ()))", self.expr(key, &exploded, true)), same_shape: false }); @@ -1621,7 +1621,7 @@ outer: { var reached : (int, ()); let mapped = ::edges | map((src, dst) -> (dst, src)); let filtered = mapped | filter((_, src) -> (src, 0) ge); - let flat = filtered | flatmap((_, src) -> src enlist) | map((dst, (_, src)) -> (src, dst)); + let flat = filtered | flatmap((k, src) -> (k, (0, src)) enlist) | map((dst, (_, src)) -> (src, dst)); let doubled = flat + flat; let once = doubled + (flat | negate); let paired = once | join(::edges, (src, dst, _) -> (src, dst)); diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs index 030fb0688..73d7b511b 100644 --- a/interactive/tests/explain.rs +++ b/interactive/tests/explain.rs @@ -440,7 +440,7 @@ fn flatmap_of_a_parent_explanations_sufficient() { let edges = input 0 : ((int, int), ()) | map(((src, dst), ()) -> ((src), (dst))); let roots = edges | map((src, _) -> (src, ())); reach: { - let hops = edges | flatmap((_, (dst)) -> (dst enlist, (dst, 1) add enlist) append) | map((src, (_, dst)) -> (src, (dst))); + let hops = edges | flatmap((k, (dst)) -> (k, (dst enlist, (dst, 1) add enlist) append) cap_list) | map((src, dst) -> (src, (dst))); let step = x | join(hops, (_, (), (dst)) -> ((dst), ())); set x = (roots + step) | distinct; } @@ -589,9 +589,20 @@ fn fuzz_explanations_sufficient() { /// op that used to `panic!` in explain. const FLATMAP_ROW: &str = r#" let rows = input 0 : ((int, int), ()) | map(((a, b), ()) -> ((a), ((a enlist, b enlist) append))); - output "result" = rows | flatmap((_, (l)) -> l); + output "result" = rows | flatmap((k, (l)) -> (k, ((0, l len) range, l) zip) cap_list); "#; +#[test] +fn flatmap_rekey_and_drop_explanations_sufficient() { + let src = r#" + let rows = input 0 : ((int, int), ()); + output "result" = rows | flatmap(((a, b), ()) -> + ((a, b), (a, b) lt) branch Result match ( + Ok (_ -> [] : List((int, int))), Err ((a, b) -> (b, a) enlist))); + "#; + assert!(!assert_all_rows_sufficient(src, &[gen_edges(20, 22)]).is_empty()); +} + #[test] fn flatmap_explanations_sufficient_small() { let sizes = assert_all_rows_sufficient(FLATMAP_ROW, &[gen_edges(20, 22)]); @@ -608,7 +619,7 @@ const LAMBDAS: &str = r#" let proposals = reach | join(edges, (_, _, dst) -> (dst, ())); set reach = roots + proposals | distinct; } - output "result" = reach::reach | map((k, _) -> (k, k)) | flatmap(((a), (b)) -> (a enlist, b enlist) append); + output "result" = reach::reach | map((k, _) -> (k, k)) | flatmap(((a), (b)) -> let l = (a enlist, b enlist) append in ((a), ((0, l len) range, l) zip) cap_list); "#; #[test] @@ -626,7 +637,7 @@ const SCALARS: &str = r#" let pairs = input 0 : (int, int); let counted = pairs | filter((_, v) -> (v, 3) gt) | count; let lists = pairs | value((_, v) -> (v, 2) mul) | collect; - let flat = lists | flatmap((_, l) -> l) | value((_, (_, x)) -> x); + let flat = lists | flatmap((k, l) -> (k, l) cap_list) | value((_, x) -> x); inner: { set low = (flat | filter((_, x) -> (x, 15) lt)) + (low | filter((k, _) -> (k, k) eq)) | min; } diff --git a/interactive/tests/flatmap.rs b/interactive/tests/flatmap.rs new file mode 100644 index 000000000..ecebb50be --- /dev/null +++ b/interactive/tests/flatmap.rs @@ -0,0 +1,92 @@ +//! Complete rows and zero-or-one lists; only the passing match lane builds output. +use interactive::{lower, parse, passes, plan, server::Server}; + +const PROGRAM: &str = r#" + type Member = User int | Group int; + let grants = input 0 : (int, (Member, int)); + output "users" = grants | flatmap((doc, (member, level)) -> + ((doc, level), member) cap_sum match ( + User (((doc, level), u) -> (u call survivor, (doc, level)) enlist), + Group (_ -> [] : List((int, (int, int)))))); + output "literal" = grants | flatmap((doc, (_, level)) -> + ((doc, level), (level, 7) eq) branch Option map_variant None (_ -> ()) + match (None (_ -> [] : List((int, int))), Some ((doc, level) -> (level call survivor, doc) enlist))); + output "all" = grants | flatmap(row -> row enlist); + output "times" = grants | flatmap(row -> row enlist) | lift; + output "empty" = grants | flatmap((doc, (_, level)) -> ((doc, level), 0) branch Result + match (Ok (_ -> [] : List((int, int))), Err (row -> row enlist))); + output "many" = grants | flatmap((doc, (_, level)) -> ((level, doc) enlist, (level, doc) enlist) append); +"#; + +struct Survivor(corgi::Shape); +impl corgi::HostKernel for Survivor { + fn name(&self) -> &str { "survivor" } + fn input(&self) -> &corgi::Shape { &self.0 } + fn output(&self) -> &corgi::Shape { &corgi::Shape::Int } + fn eval(&self, input: corgi::Value) -> Result { + let input = input.into_prod("survivor arguments")?.remove(0); + assert!(input.as_i64("survivor")?.iter().all(|&x| x != 0), "built a dropped row"); + Ok(input) + } +} +fn register() { + static ONCE: std::sync::Once = std::sync::Once::new(); + ONCE.call_once(|| interactive::lambda::register(std::sync::Arc::new(Survivor(corgi::Shape::Prod(vec![corgi::Shape::Int]))))); +} + +#[test] +fn lanes_diffs_retractions_and_times() { + register(); + for workers in [1, 3] { + timely::execute(timely::Config::process(workers), |worker| { + let mut program = lower::lower_tree(parse::pipe::parse(PROGRAM)); + passes::optimize(&mut program); + passes::projection::pushdown(&mut program, &|_| None).unwrap(); + let mut server = Server::new(); + server.set_export_taps(true); + server.install(worker, "p", &program).unwrap(); + for (epoch, diff) in [3, -2, -1].into_iter().enumerate() { + if worker.index() == 0 { + server.feed("p", 0, &format!( + "10 val=tuple(inject(0,2),7) diff={diff}\n20 val=tuple(inject(1,0),0) diff={diff}"), None).unwrap(); + } + server.tick(worker); + let count = [3, 1, 0][epoch]; + for (name, k, v) in [("users", "2", "tuple(10,7)"), ("literal", "7", "10")] { + let got = server.snapshot(worker, &format!("p::{name}")).unwrap().rows(); + let want = if count == 0 || worker.index() != 0 { vec![] } else { vec![(k.into(), v.into(), count)] }; + assert_eq!(got, want, "{name}, epoch {epoch}"); + let changes = server.take_changes(&format!("p::{name}")); + assert!(changes.iter().all(|c| c.times.to_vec().iter().all(|t| + *t == interactive::server::host_time(epoch as u64)))); + } + assert!(server.snapshot(worker, "p::empty").unwrap().rows().is_empty()); + assert_eq!(server.snapshot(worker, "p::all").unwrap().rows().len(), if count == 0 || worker.index() != 0 { 0 } else { 2 }); + let many = server.snapshot(worker, "p::many").unwrap().rows(); + assert!(many.iter().all(|r| r.2 == 2 * count)); + } + }).unwrap().join().into_iter().for_each(|r| r.unwrap()); + } +} + +#[test] +fn printers_roundtrip_named_arms() { + register(); + let config = passes::Config::parse("all").unwrap(); + let report = plan::report_at(PROGRAM, &|_| None, &config, 0).unwrap(); + let text = report.emit_ddp().unwrap().text; + assert!(text.contains("flatmap("), "{text}"); + let again = plan::report_at(&text, &|_| None, &config, 0).unwrap(); + assert!(report.compare_final(&again).is_empty()); +} + +#[test] +fn malformed_rows_are_rejected() { + timely::execute_directly(|worker| { + for expr in ["flatmap(row -> row)", "flatmap(_ -> 1 enlist)"] { + let p = lower::lower_tree(parse::pipe::parse(&format!( + "output \"x\" = input 0 : (int, ()) | {expr};"))); + assert!(Server::new().install(worker, "p", &p).is_err(), "{expr}"); + } + }); +} diff --git a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.ddp b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.ddp index 8e51eb164..2bc5dbf92 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.ddp +++ b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.ddp @@ -33,7 +33,7 @@ let jobs_join3 = jobs_idx3 | join(place_idx, (_, (key, (company, country)), (the let jobs = jobs_join3 | filter((_, (_, country, there)) -> (country, there) eq) | map((key, (company, _, _)) -> (key, company)); let jobs_idx4 = jobs | arrange; let named_reduce = jobs_idx4 | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); -let ranked_rows = named_reduce | flatmap((_, entries) -> entries) | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))); +let ranked_rows = named_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list) | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))); let ranked_idx = ranked_rows | arrange; let ranked_join = ranked_idx | join(person_idx, (node, (rid, since, pos, company), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since, node, pos neg), (first, last, company)))); let ranked_idx2 = ranked_join | arrange; diff --git a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.lowered--optimized.diff b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.lowered--optimized.diff index 4812e9483..05849ae6a 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.lowered--optimized.diff +++ b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.lowered--optimized.diff @@ -25,12 +25,12 @@ +let jobs = jobs_join3 | filter((_, (_, country, there)) -> (country, there) eq) | map((key, (company, _, _)) -> (key, company)); let jobs_idx4 = jobs | arrange; let named_reduce = jobs_idx4 | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); --let named = named_reduce | flatmap((_, entries) -> entries); +-let named = named_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); -let ranked_map = named | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))); -let ranked_idx = ranked_map | arrange; -let person_idx2 = person | arrange; -let ranked_join = ranked_idx | join(person_idx2, (node, (rid, since, pos, company), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since, node, pos neg), (first, last, company)))); -+let ranked_rows = named_reduce | flatmap((_, entries) -> entries) | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))); ++let ranked_rows = named_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list) | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))); +let ranked_idx = ranked_rows | arrange; +let ranked_join = ranked_idx | join(person_idx, (node, (rid, since, pos, company), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since, node, pos neg), (first, last, company)))); let ranked_idx2 = ranked_join | arrange; diff --git a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed--lowered.diff b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed--lowered.diff index 1ff2a9cf0..d496f1c06 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed--lowered.diff +++ b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed--lowered.diff @@ -13,7 +13,7 @@ -let hop2 = hop1 | join(friend, (_, req, (g, _)) -> (g, req)); -let circle = hop1 + hop2 | filter((node, (_, who, _, _)) -> (node, who) ne) | map((node, (rid, _, country, year)) -> ((node, rid, country, year), ())) | map((k, v) -> ((k, v), ())) | reduce((_, vals) -> vals map (((), _) -> ((), 1))) | map((kv, ()) -> kv); -let jobs = circle | map(((node, rid, country, year), ()) -> (node, (rid, country, year))) | join(work, (node, (rid, country, year), (at, since)) -> (at, (rid, node, country, year, since))) | filter((_, (_, _, _, year, since)) -> (since, year) lt) | join(org, (_, (rid, node, country, _, since), (company, _, located)) -> (located, ((rid, since, node), (company, country)))) | join(place, (_, (key, (company, country)), (there, _, _)) -> (key, (company, country, there))) | filter((_, (_, country, there)) -> (country, there) eq) | map((key, (company, _, _)) -> (key, company)); --let named = jobs | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((_, entries) -> entries); +-let named = jobs | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); -let ranked = named | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))) | join(person, (node, (rid, since, pos, company), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since, node, pos neg), (first, last, company)))) | reduce((_, vals) -> let positive = (vals map ((_, d) -> (d, 0) gt), (0, vals len) range) zip filter in let indices = ((0, (positive len, 10) min) range, positive) gather in let prefix = (indices, vals) gather in let ends = prefix map ((_, d) -> (d, 10) min) scan_add in let chosen = (prefix, ends) zip map (((item, d), end) -> let count = (d, 10) min in let before = (end, count) sub in let keep = (((10, before) sub, 0) max, count) min in ((keep, 0) gt, (item, keep))) filter in let ends = chosen map ((_, d) -> d) scan_add in (chosen, ends) zip map (((item, d), end) -> (item, ((end, d) sub, end) range) cap_list map ((item, rank) -> ((rank, item), 1))) flatten .1); +let hop1_map = request | map(((rid, who, country, year), ()) -> (who, (rid, who, country, year))); +let hop1_idx = hop1_map | arrange; @@ -45,7 +45,7 @@ +let jobs = jobs_filter2 | map((key, (company, _, _)) -> (key, company)); +let jobs_idx4 = jobs | arrange; +let named_reduce = jobs_idx4 | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); -+let named = named_reduce | flatmap((_, entries) -> entries); ++let named = named_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); +let ranked_map = named | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))); +let ranked_idx = ranked_map | arrange; +let person_idx2 = person | arrange; diff --git a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed.ddp b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed.ddp index 1325bd952..3790672dc 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed.ddp +++ b/interactive/tests/plan_pairs/expected/ldbc_ic11_oneoff.parsed.ddp @@ -8,7 +8,7 @@ let hop1 = request | map(((rid, who, country, year), ()) -> (who, (rid, who, cou let hop2 = hop1 | join(friend, (_, req, (g, _)) -> (g, req)); let circle = hop1 + hop2 | filter((node, (_, who, _, _)) -> (node, who) ne) | map((node, (rid, _, country, year)) -> ((node, rid, country, year), ())) | map((k, v) -> ((k, v), ())) | reduce((_, vals) -> vals map (((), _) -> ((), 1))) | map((kv, ()) -> kv); let jobs = circle | map(((node, rid, country, year), ()) -> (node, (rid, country, year))) | join(work, (node, (rid, country, year), (at, since)) -> (at, (rid, node, country, year, since))) | filter((_, (_, _, _, year, since)) -> (since, year) lt) | join(org, (_, (rid, node, country, _, since), (company, _, located)) -> (located, ((rid, since, node), (company, country)))) | join(place, (_, (key, (company, country)), (there, _, _)) -> (key, (company, country, there))) | filter((_, (_, country, there)) -> (country, there) eq) | map((key, (company, _, _)) -> (key, company)); -let named = jobs | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((_, entries) -> entries); +let named = jobs | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let ranked = named | map(((rid, since, node), (pos, company)) -> (node, (rid, since, pos, company))) | join(person, (node, (rid, since, pos, company), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since, node, pos neg), (first, last, company)))) | reduce((_, vals) -> let positive = (vals map ((_, d) -> (d, 0) gt), (0, vals len) range) zip filter in let indices = ((0, (positive len, 10) min) range, positive) gather in let prefix = (indices, vals) gather in let ends = prefix map ((_, d) -> (d, 10) min) scan_add in let chosen = (prefix, ends) zip map (((item, d), end) -> let count = (d, 10) min in let before = (end, count) sub in let keep = (((10, before) sub, 0) max, count) min in ((keep, 0) gt, (item, keep))) filter in let ends = chosen map ((_, d) -> d) scan_add in (chosen, ends) zip map (((item, d), end) -> (item, ((end, d) sub, end) range) cap_list map ((item, rank) -> ((rank, item), 1))) flatten .1); let answer = ranked | map((rid, (rank, ((since, node, _), (first, last, company)))) -> ((rid, rank), (node, first, last, company, since))); output "answer" = answer | arrange; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is03.ddp b/interactive/tests/plan_pairs/expected/ldbc_is03.ddp index 0d5b730db..004a3bc59 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is03.ddp +++ b/interactive/tests/plan_pairs/expected/ldbc_is03.ddp @@ -12,7 +12,7 @@ let person_idx = person | arrange; let ranked_join2 = ranked_idx2 | join(person_idx, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))); let ranked_idx3 = ranked_join2 | arrange; let ranked_reduce = ranked_idx3 | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); -let answer = ranked_reduce | flatmap((_, entries) -> entries) | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); +let answer = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list) | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); let answer_idx = answer | arrange; -- Renderer: publish the existing arrangement. output "answer" = answer_idx; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is03.lowered--optimized.diff b/interactive/tests/plan_pairs/expected/ldbc_is03.lowered--optimized.diff index e85652ae2..1dac17235 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is03.lowered--optimized.diff +++ b/interactive/tests/plan_pairs/expected/ldbc_is03.lowered--optimized.diff @@ -4,9 +4,9 @@ let ranked_join2 = ranked_idx2 | join(person_idx, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))); let ranked_idx3 = ranked_join2 | arrange; let ranked_reduce = ranked_idx3 | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); --let ranked = ranked_reduce | flatmap((_, entries) -> entries); +-let ranked = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); -let answer = ranked | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); -+let answer = ranked_reduce | flatmap((_, entries) -> entries) | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); ++let answer = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list) | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); let answer_idx = answer | arrange; -- Renderer: publish the existing arrangement. output "answer" = answer_idx; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is03.parsed--lowered.diff b/interactive/tests/plan_pairs/expected/ldbc_is03.parsed--lowered.diff index c849e10e7..c5858c79f 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is03.parsed--lowered.diff +++ b/interactive/tests/plan_pairs/expected/ldbc_is03.parsed--lowered.diff @@ -6,7 +6,7 @@ let request = input 0 : ((int, int), ()); let person = import "graph::person" : (int, (List(int), List(int), List(int), int, int, List(int), List(int), int, int, int, int)); let friend = import "graph::friend" : (int, (int, int)); --let ranked = request | map(((rid, who), ()) -> (who, rid)) | join(friend, (_, rid, (f, since)) -> (f, (rid, since))) | join(person, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((_, entries) -> entries); +-let ranked = request | map(((rid, who), ()) -> (who, rid)) | join(friend, (_, rid, (f, since)) -> (f, (rid, since))) | join(person, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); +let ranked_map = request | map(((rid, who), ()) -> (who, rid)); +let ranked_idx = ranked_map | arrange; +let friend_idx = friend | arrange; @@ -16,7 +16,7 @@ +let ranked_join2 = ranked_idx2 | join(person_idx, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))); +let ranked_idx3 = ranked_join2 | arrange; +let ranked_reduce = ranked_idx3 | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); -+let ranked = ranked_reduce | flatmap((_, entries) -> entries); ++let ranked = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); -output "answer" = answer | arrange; +let answer_idx = answer | arrange; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is03.parsed.ddp b/interactive/tests/plan_pairs/expected/ldbc_is03.parsed.ddp index 8afff5cf2..0f8130223 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is03.parsed.ddp +++ b/interactive/tests/plan_pairs/expected/ldbc_is03.parsed.ddp @@ -1,6 +1,6 @@ let request = input 0 : ((int, int), ()); let person = import "graph::person" : (int, (List(int), List(int), List(int), int, int, List(int), List(int), int, int, int, int)); let friend = import "graph::friend" : (int, (int, int)); -let ranked = request | map(((rid, who), ()) -> (who, rid)) | join(friend, (_, rid, (f, since)) -> (f, (rid, since))) | join(person, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((_, entries) -> entries); +let ranked = request | map(((rid, who), ()) -> (who, rid)) | join(friend, (_, rid, (f, since)) -> (f, (rid, since))) | join(person, (f, (rid, since), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((since neg, f), (first, last)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((neg, f), (first, last)))) -> ((rid, rank), (f, first, last, neg neg))); output "answer" = answer | arrange; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is07.ddp b/interactive/tests/plan_pairs/expected/ldbc_is07.ddp index c1d6c532a..2d4c91276 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is07.ddp +++ b/interactive/tests/plan_pairs/expected/ldbc_is07.ddp @@ -34,7 +34,7 @@ let person_idx = person | arrange; let ranked_join = marked_idx | join(person_idx, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))); let ranked_idx = ranked_join | arrange; let ranked_reduce = ranked_idx | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); -let answer = ranked_reduce | flatmap((_, entries) -> entries) | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); +let answer = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list) | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); let answer_idx = answer | arrange; -- Renderer: publish the existing arrangement. output "answer" = answer_idx; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is07.lowered--optimized.diff b/interactive/tests/plan_pairs/expected/ldbc_is07.lowered--optimized.diff index 1019d56ba..95e6ffd32 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is07.lowered--optimized.diff +++ b/interactive/tests/plan_pairs/expected/ldbc_is07.lowered--optimized.diff @@ -26,9 +26,9 @@ let ranked_join = marked_idx | join(person_idx, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))); let ranked_idx = ranked_join | arrange; let ranked_reduce = ranked_idx | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); --let ranked = ranked_reduce | flatmap((_, entries) -> entries); +-let ranked = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); -let answer = ranked | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); -+let answer = ranked_reduce | flatmap((_, entries) -> entries) | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); ++let answer = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list) | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); let answer_idx = answer | arrange; -- Renderer: publish the existing arrangement. output "answer" = answer_idx; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is07.parsed--lowered.diff b/interactive/tests/plan_pairs/expected/ldbc_is07.parsed--lowered.diff index 344585511..fe232509a 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is07.parsed--lowered.diff +++ b/interactive/tests/plan_pairs/expected/ldbc_is07.parsed--lowered.diff @@ -11,7 +11,7 @@ -let reply = request | map(((rid, m), ()) -> (m, rid)) | join(message, (m, rid, (_, _, op, _, _, _, _, _, _, _, _, _, _)) -> (m, (rid, op))) | join(by_parent, (_, (rid, op), id) -> (id, (rid, op))) | join(message, (id, (rid, op), (kind, created, who, _, text, _, _, _, _, _, _, _, _)) -> (kind, ((who, op), (rid, id, text, created)))) | filter((kind, _) -> (kind, 1) eq) | map((_, row) -> row); -let known = reply | map(((who, op), row) -> (who, (op, row))) | join(friend, (who, (op, row), (f, _)) -> ((f, op) eq, ((who, op), row))) | filter((same, _) -> same) | map((_, (pair, row)) -> ((pair, row), ())) | map((k, v) -> ((k, v), ())) | reduce((_, vals) -> vals map (((), _) -> ((), 1))) | map((kv, ()) -> kv) | map(((pair, row), ()) -> (pair, row)); -let marked = (known | map(((who, _), row) -> (who, (row, 1)))) + (reply + (known | negate) | map(((who, _), row) -> (who, (row, 0)))); --let ranked = marked | join(person, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((_, entries) -> entries); +-let ranked = marked | join(person, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); +let reply_map = request | map(((rid, m), ()) -> (m, rid)); +let reply_idx = reply_map | arrange; +let message_idx = message | arrange; @@ -43,7 +43,7 @@ +let ranked_join = marked_idx | join(person_idx, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))); +let ranked_idx = ranked_join | arrange; +let ranked_reduce = ranked_idx | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist); -+let ranked = ranked_reduce | flatmap((_, entries) -> entries); ++let ranked = ranked_reduce | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); -output "answer" = answer | arrange; +let answer_idx = answer | arrange; diff --git a/interactive/tests/plan_pairs/expected/ldbc_is07.parsed.ddp b/interactive/tests/plan_pairs/expected/ldbc_is07.parsed.ddp index b7f84a0c9..772f356c7 100644 --- a/interactive/tests/plan_pairs/expected/ldbc_is07.parsed.ddp +++ b/interactive/tests/plan_pairs/expected/ldbc_is07.parsed.ddp @@ -6,6 +6,6 @@ let friend = import "graph::friend" : (int, (int, int)); let reply = request | map(((rid, m), ()) -> (m, rid)) | join(message, (m, rid, (_, _, op, _, _, _, _, _, _, _, _, _, _)) -> (m, (rid, op))) | join(by_parent, (_, (rid, op), id) -> (id, (rid, op))) | join(message, (id, (rid, op), (kind, created, who, _, text, _, _, _, _, _, _, _, _)) -> (kind, ((who, op), (rid, id, text, created)))) | filter((kind, _) -> (kind, 1) eq) | map((_, row) -> row); let known = reply | map(((who, op), row) -> (who, (op, row))) | join(friend, (who, (op, row), (f, _)) -> ((f, op) eq, ((who, op), row))) | filter((same, _) -> same) | map((_, (pair, row)) -> ((pair, row), ())) | map((k, v) -> ((k, v), ())) | reduce((_, vals) -> vals map (((), _) -> ((), 1))) | map((kv, ()) -> kv) | map(((pair, row), ()) -> (pair, row)); let marked = (known | map(((who, _), row) -> (who, (row, 1)))) + (reply + (known | negate) | map(((who, _), row) -> (who, (row, 0)))); -let ranked = marked | join(person, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((_, entries) -> entries); +let ranked = marked | join(person, (who, ((rid, id, text, created), k), (first, last, _, _, _, _, _, _, _, _, _)) -> (rid, ((created neg, who), (id, text, first, last, k)))) | reduce((_, vals) -> let lists = vals map ((v, d) -> (v, (0, d) range) cap_list map ((v, _) -> v)) in let flat = lists flatten in (flat.1, 1) enlist) | flatmap((k, entries) -> (k, ((0, entries len) range, entries) zip) cap_list); let answer = ranked | map((rid, (rank, ((neg, who), (id, text, first, last, k)))) -> ((rid, rank), (id, text, neg neg, who, first, last, k))); output "answer" = answer | arrange; diff --git a/interactive/tests/plan_pairs/expected/server_arrange_idem.ddp b/interactive/tests/plan_pairs/expected/server_arrange_idem.ddp index 8d2f11fbe..933b2d63e 100644 --- a/interactive/tests/plan_pairs/expected/server_arrange_idem.ddp +++ b/interactive/tests/plan_pairs/expected/server_arrange_idem.ddp @@ -2,7 +2,7 @@ let input0 = input 0 : ((int), (int, List(int))); let node = input0 | map(((id), v) -> (id, v)); let elems = node | map((id, _) -> (id, id)); -let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((_, cs) -> cs) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); +let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((k, cs) -> (k, ((0, cs len) range, cs) zip) cap_list) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); let pc_idx = pc | arrange; let producing_filter = node | filter((_, (op, _)) -> (op, 1) eq); let producing_map = producing_filter | map((id, _) -> (id, ())); diff --git a/interactive/tests/plan_pairs/expected/server_arrange_idem.lowered--optimized.diff b/interactive/tests/plan_pairs/expected/server_arrange_idem.lowered--optimized.diff index ed9679db2..50c0497e9 100644 --- a/interactive/tests/plan_pairs/expected/server_arrange_idem.lowered--optimized.diff +++ b/interactive/tests/plan_pairs/expected/server_arrange_idem.lowered--optimized.diff @@ -5,9 +5,9 @@ let node = input0 | map(((id), v) -> (id, v)); let elems = node | map((id, _) -> (id, id)); -let pc_map = node | map((id, (op, cs)) -> ((id, op), cs)); --let pc_rows = pc_map | flatmap((_, cs) -> cs); +-let pc_rows = pc_map | flatmap((k, cs) -> (k, ((0, cs len) range, cs) zip) cap_list); -let pc = pc_rows | map(((id, op), (pos, child)) -> (child, (id, op, pos))); -+let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((_, cs) -> cs) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); ++let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((k, cs) -> (k, ((0, cs len) range, cs) zip) cap_list) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); +let pc_idx = pc | arrange; let producing_filter = node | filter((_, (op, _)) -> (op, 1) eq); let producing_map = producing_filter | map((id, _) -> (id, ())); diff --git a/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed--lowered.diff b/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed--lowered.diff index 9eadd917f..c0b2ccf6b 100644 --- a/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed--lowered.diff +++ b/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed--lowered.diff @@ -6,11 +6,11 @@ +let input0 = input 0 : ((int), (int, List(int))); +let node = input0 | map(((id), v) -> (id, v)); let elems = node | map((id, _) -> (id, id)); --let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((_, cs) -> cs) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); +-let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((k, cs) -> (k, ((0, cs len) range, cs) zip) cap_list) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); -let producing = (node | filter((_, (op, _)) -> (op, 1) eq) | map((id, _) -> (id, ()))) + (node | filter((_, (op, _)) -> (op, 2) eq) | map((id, _) -> (id, ()))); -let arr = node | filter((_, (op, _)) -> (op, 1) eq) | map((id, (_, cs)) -> (id, (0, cs) get)); +let pc_map = node | map((id, (op, cs)) -> ((id, op), cs)); -+let pc_rows = pc_map | flatmap((_, cs) -> cs); ++let pc_rows = pc_map | flatmap((k, cs) -> (k, ((0, cs len) range, cs) zip) cap_list); +let pc = pc_rows | map(((id, op), (pos, child)) -> (child, (id, op, pos))); +let producing_filter = node | filter((_, (op, _)) -> (op, 1) eq); +let producing_map = producing_filter | map((id, _) -> (id, ())); diff --git a/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed.ddp b/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed.ddp index 50738357f..fa2b5313a 100644 --- a/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed.ddp +++ b/interactive/tests/plan_pairs/expected/server_arrange_idem.parsed.ddp @@ -1,6 +1,6 @@ let node = input 0 : ((int), (int, List(int))) | map(((id), v) -> (id, v)); let elems = node | map((id, _) -> (id, id)); -let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((_, cs) -> cs) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); +let pc = node | map((id, (op, cs)) -> ((id, op), cs)) | flatmap((k, cs) -> (k, ((0, cs len) range, cs) zip) cap_list) | map(((id, op), (pos, child)) -> (child, (id, op, pos))); let producing = (node | filter((_, (op, _)) -> (op, 1) eq) | map((id, _) -> (id, ()))) + (node | filter((_, (op, _)) -> (op, 2) eq) | map((id, _) -> (id, ()))); let arr = node | filter((_, (op, _)) -> (op, 1) eq) | map((id, (_, cs)) -> (id, (0, cs) get)); eqsat: { diff --git a/interactive/tests/programs/f64_math.ddp b/interactive/tests/programs/f64_math.ddp index 066f4acbd..1300a37f4 100644 --- a/interactive/tests/programs/f64_math.ddp +++ b/interactive/tests/programs/f64_math.ddp @@ -32,7 +32,7 @@ let specials = xs -- A row-at-a-time filter and flatmap. let big = xs | filter((_, (x, _)) -> (x call fexp, 2.0) call fgt); -let spread = xs | flatmap((_, (x, y)) -> (x call fabs call fsqrt enlist, y call fexp enlist) append); +let spread = xs | flatmap((k, (x, y)) -> let l = (x call fabs call fsqrt enlist, y call fexp enlist) append in (k, ((0, l len) range, l) zip) cap_list); -- A join whose projection runs a row at a time. let scale = input 1 : ((int, int), ()) | map(((a, b), ()) -> (a, b)); diff --git a/interactive/tests/programs/lambda_slots.ddp b/interactive/tests/programs/lambda_slots.ddp index 23ac7854c..f86da0347 100644 --- a/interactive/tests/programs/lambda_slots.ddp +++ b/interactive/tests/programs/lambda_slots.ddp @@ -1,5 +1,5 @@ -- The slots that take one lambda: `value` (the value as written, with no tuple around it), --- `filter`, `flatmap` (a row per element, valued `(position, element)`) and `delay` (a row +-- `filter`, `flatmap` (complete rows, with positions explicitly zipped here) and `delay` (a row -- advances to the coordinate-wise maximum of its time and the tuple it names). A `--` comment -- may sit inside a lambda written over several lines. Declared types and the built-in `Option` -- are enums a lambda builds and takes apart, also when a payload is itself a sum. @@ -12,7 +12,7 @@ let pairs = input 0 : ((int, int), ()) | map(((a, b), ()) -> ((a), (b))); output "value" = pairs | value(((a), (b)) -> ((b, 10) mul, a) add); output "filter" = pairs | filter(((a), (b)) -> (a, b) lt); -output "flatmap" = pairs | flatmap(((a), (b)) -> (a enlist, b enlist, (a, b) add enlist) append); +output "flatmap" = pairs | flatmap(row -> let ((a), (b)) = row in let l = (a enlist, b enlist, (a, b) add enlist) append in (row.0, ((0, l len) range, l) zip) cap_list); stepped: { let delayed = pairs | delay((_, (b)) -> (0, (256, b call bit_length) mul)); set counted = delayed | count; diff --git a/interactive/tests/programs/registered.ddp b/interactive/tests/programs/registered.ddp index 7e2819436..83c0ba2bb 100644 --- a/interactive/tests/programs/registered.ddp +++ b/interactive/tests/programs/registered.ddp @@ -11,7 +11,7 @@ let seeds = input 0 : ((int), ()) | map(((n), ()) -> (n, ())); -- Refine: each cell's children are `cell call grow`, until `grow` returns none. refine: { - let kids = cells | flatmap((n, ()) -> n call grow) | map((_, (_, (kid, _))) -> (kid, ())); + let kids = cells | flatmap((n, ()) -> n call grow map ((kid, _) -> (kid, ()))); set cells = seeds + kids | distinct; } let cells = refine::cells;