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).
 
 </details>
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<usize> = 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<Graph<NumOp>> {
     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<Expr>),
-    /// 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<Expr>, Lambda),
     /// Pair each row's value with its whole time. See `LinearOp::Lift`.
     Lift(Box<Expr>),
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<Shape, String> {
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<corgi::Value, String> {
+        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;