Answer a fold in a column from one pass over its list

A fold in a table's column runs once per row, over its whole list each
time: per branch, the sum of its accounts read all 100,000 accounts for
every branch, and per rich account, how many are richer read them all for
each of 404 accounts. At load, a planned fold in a column whose test
compares a field of its item with a key the row gives, the rest of it
reading only its list, its item and constants, gets a shape: a group (the
field equals the key) or a rank (a count of items whose field orders
against the key, and nothing else). The app then keeps one pass over the
list per fold and list version: a group pass runs the block a leaf at a
time with the comparison taken as true and the key left out, folding each
item into its field's group; a rank pass radix-sorts the field's values.
A row runs its key alone and looks it up. The pass is made at the second
row that asks, so a fold asked once scans once.

Answers are the fold's own: items join their group in order, so a sum
adds as it would; a key that faults leaves that row to run the fold as
written, and an item that would fault in the pass leaves every row to.
lane_folds now also makes shaped folds (185 groups, 127 ranks, 98 passes
made) and checks them against the same module run item by item.

On the bank of 100,000 accounts: per-branch totals 4.8 ms (16 ms before,
SQLite 46 ms, 14 ms with the index a DBA would add on branch and balance)
and each rich account's rank 6.7 ms (268 ms before, SQLite 2.84 s, 7.6 ms
with an index on balance), HTTP and JSON included on Pure's side. Pure is
given no index.

5 files changed, 474 insertions, 17 deletions

diff --git a/core/module-list.c b/core/module-list.c
index 55c6812..5df66f4 100644
--- a/core/module-list.c
+++ b/core/module-list.c
@@ -510,6 +510,11 @@ bool pure_list_current(const PureList *copy, const PureList *list) {
     return copy->id == list->id && copy->version == list->version;
 }
 
+void pure_list_stamp(const PureList *list, uint64_t *id, uint64_t *version) {
+    *id = list->id;
+    *version = list->version;
+}
+
 static bool log_change(PureUndo *undo, Entry entry) {
     if (undo->count == undo->capacity) {
         size_t capacity = undo->capacity ? 2 * undo->capacity : 16;
diff --git a/core/module-list.h b/core/module-list.h
index e370b97..5e724b8 100644
--- a/core/module-list.h
+++ b/core/module-list.h
@@ -60,6 +60,8 @@ PureList *pure_list_share(PureList *, PureRetired *retired);
 void pure_list_hold(PureList *copy);
 /* Whether copy shares list's items as they are now. */
 bool pure_list_current(const PureList *copy, const PureList *list);
+/* Which list, and its version: the same pair means the same items. */
+void pure_list_stamp(const PureList *, uint64_t *id, uint64_t *version);
 /* Free what was retired while no copy that shared it remains. */
 void pure_retired_free(PureRetired *);
 /* Insert a copy of cells (text bytes are copied) before position at. A NULL
diff --git a/core/module.c b/core/module.c
index 36e13c3..07d98d4 100644
--- a/core/module.c
+++ b/core/module.c
@@ -433,6 +433,128 @@ PURE_COLD static void plan_folds(PureModule *m, PureBlock block) {
     }
 }
 
+/* A planned fold's block as a tree (its lane forms have no branch; one
+ * with PICK or SLIDE is left alone): per instruction, its operands' nodes
+ * and the first instruction of its subtree, whether it reads the fold's
+ * item (item), and whether it reads nothing but the fold's list, its item
+ * and constants (own): what one pass over the list can compute. */
+typedef struct { uint32_t first; int kid[3]; bool item, own; } Node;
+
+PURE_COLD static unsigned arity(uint8_t op) {
+    switch (op) {
+    case PURE_OP_CONST: case PURE_OP_LOAD: case PURE_OP_INDEX_LOAD: case PURE_OP_ROW: case PURE_OP_OUTER_ROW:
+    case PURE_OP_LIST_COUNT: case PURE_OP_INDEX_FIRST: case PURE_OP_INDEX_LAST:
+        return 0;
+    case PURE_OP_ITEM_GET: case PURE_OP_LIST_FIND: case PURE_OP_LIST_HAS: case PURE_OP_NEG: case PURE_OP_NOT:
+    case PURE_OP_FLOOR: case PURE_OP_INDEX_NUM:
+        return 1;
+    case PURE_OP_CLAMP: return 3;
+    default: return 2;
+    }
+}
+
+/* The list an instruction reads, if any (-1: none). */
+PURE_COLD static int list_read(const PureOp *op) {
+    switch (op->op) {
+    case PURE_OP_ITEM_GET: return (int)list_of(op->arg);
+    case PURE_OP_LIST_COUNT: case PURE_OP_INDEX_FIRST: case PURE_OP_INDEX_LAST: case PURE_OP_LIST_FIND:
+    case PURE_OP_LIST_HAS: case PURE_OP_ROW:
+        return (int)op->arg;
+    default: return -1;
+    }
+}
+
+PURE_COLD static bool block_tree(const PureModule *m, PureBlock block, int list, Node *nodes, int *roots, unsigned *count) {
+    int stack[PURE_MODULE_STACK];
+    unsigned sp = 0;
+    for (uint32_t i = 0; i < block.length; ++i) {
+        const PureOp *op = &m->code[block.start + i];
+        if (op->op == PURE_OP_PICK || op->op == PURE_OP_SLIDE) return false;
+        unsigned pops = arity(op->op);
+        Node *node = &nodes[i];
+        int read = list_read(op);
+        node->first = block.start + i;
+        node->item = op->op == PURE_OP_ROW;
+        node->own = op->op != PURE_OP_LOAD && op->op != PURE_OP_INDEX_LOAD && op->op != PURE_OP_OUTER_ROW &&
+                    (read < 0 || read == list);
+        for (unsigned k = 0; k < 3; ++k) node->kid[k] = -1;
+        for (unsigned k = 0; k < pops; ++k) {
+            int kid = stack[sp - pops + k];
+            node->kid[k] = kid;
+            node->item = node->item || nodes[kid].item;
+            node->own = node->own && nodes[kid].own;
+        }
+        if (pops) node->first = nodes[stack[sp - pops]].first;
+        sp -= pops;
+        stack[sp++] = (int)i;
+    }
+    memcpy(roots, stack, sp * sizeof(*stack));
+    *count = sp;
+    return true;
+}
+
+/* The comparisons a test ANDs together (at most 16). */
+PURE_COLD static bool conjuncts(const PureModule *m, PureBlock block, const Node *nodes, int at, int *out, unsigned *count) {
+    if (m->code[block.start + at].op == PURE_OP_AND)
+        return conjuncts(m, block, nodes, nodes[at].kid[0], out, count) &&
+               conjuncts(m, block, nodes, nodes[at].kid[1], out, count);
+    if (*count == 16) return false;
+    out[(*count)++] = at;
+    return true;
+}
+
+/* The shape of a planned fold in a column, if it has one (see PureShape). */
+PURE_COLD static void shape_fold(PureModule *m, uint32_t pc) {
+    PureOp *fold = &m->code[pc];
+    unsigned how = fold->arg & 0xFF;
+    int list = (int)(fold->arg >> 8 & 0xFF);
+    PureBlock block = {pc + 1, fold->arg >> 16};
+    if (!fold->plan || m->shapes == PURE_MODULE_SHAPES) return;
+    Node *nodes = malloc(block.length * sizeof(*nodes));
+    int roots[PURE_MODULE_STACK], tests[16];
+    unsigned count = 0, tested = 0;
+    if (!nodes) return;
+    if (!block_tree(m, block, list, nodes, roots, &count) || !conjuncts(m, block, nodes, roots[count - 1], tests, &tested) ||
+        (count == 2 && !nodes[roots[0]].own)) {
+        free(nodes);
+        return;
+    }
+    /* One comparison of the item's field with a key; the rest the item's own. */
+    int test = -1;
+    for (unsigned k = 0; k < tested; ++k) {
+        if (nodes[tests[k]].own) continue;
+        if (test >= 0) test = -2;
+        else test = tests[k];
+    }
+    PureShape shape = {0};
+    if (test >= 0) {
+        const PureOp *compare = &m->code[block.start + test];
+        const Node *node = &nodes[test];
+        uint8_t op = compare->op;
+        for (unsigned side = 0; side < 2 && !shape.form && (op == PURE_OP_EQ || op == PURE_OP_LT || op == PURE_OP_LE ||
+                                                              op == PURE_OP_GT || op == PURE_OP_GE); ++side) {
+            int field = node->kid[side], key = node->kid[1 - side];
+            const PureOp *get = &m->code[block.start + field];
+            if (get->op != PURE_OP_ITEM_GET || m->code[block.start + nodes[field].kid[0]].op != PURE_OP_ROW ||
+                nodes[key].item)
+                continue;
+            static const uint8_t flipped[] = {[PURE_OP_LT] = PURE_OP_GT, [PURE_OP_LE] = PURE_OP_GE,
+                                              [PURE_OP_GT] = PURE_OP_LT, [PURE_OP_GE] = PURE_OP_LE};
+            shape = (PureShape){op == PURE_OP_EQ ? PURE_SHAPE_GROUP : PURE_SHAPE_RANK,
+                                op == PURE_OP_EQ || !side ? op : flipped[op], (uint16_t)item_field(get->arg),
+                                block.start + (uint32_t)test,
+                                {nodes[key].first, block.start + (uint32_t)key + 1 - nodes[key].first}};
+        }
+        /* A rank answers a count tested by the comparison alone. */
+        if (shape.form == PURE_SHAPE_RANK && (how != PURE_FOLD_COUNT || tested != 1)) shape.form = 0;
+    }
+    if (shape.form) {
+        m->shape[m->shapes++] = shape;
+        fold->shape = m->shapes;
+    }
+    free(nodes);
+}
+
 PURE_COLD PureResult pure_module_read(const char *path, PureModule *out, char *error, size_t error_size) {
     FILE *file = fopen(path, "rb");
     unsigned char *bytes = NULL;
@@ -706,6 +828,12 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule
         plan_folds(out, out->view_code[i]);
         out->view_plan[i] = out->view_type[i] == PURE_VIEW_TABLE && out->view_code[i].length &&
                             block_has_lanes(out, out->view_code[i]);
+        /* A fold in a column runs for every row: one pass may answer them all. */
+        for (uint32_t pc = out->view_code[i].start; out->view_table[i] && pc < out->view_code[i].start + out->view_code[i].length; ++pc)
+            if (out->code[pc].op == PURE_OP_FOLD) {
+                shape_fold(out, pc);
+                pc += out->code[pc].arg >> 16;
+            }
     }
     for (uint32_t i = 0; i < out->handlers; ++i) plan_folds(out, out->handler[i].code);
     return PURE_OK;
@@ -1053,7 +1181,7 @@ static bool lanes_op(uint8_t op, const double *a, size_t sa, const double *b, si
  * per stack depth, or into the list's leaf). false when run would fault for
  * some item. */
 static bool run_lanes(const PureModule *m, const double *values, PureBlock block, const Event *ev, const PureRun *items,
-                      Lanes *result, double (*store)[PURE_LIST_RUN], Memo *memo) {
+                      Lanes *result, double (*store)[PURE_LIST_RUN], Memo *memo, const PureShape *shape) {
     PureApp *app = ev->app;
     Lanes stack[PURE_MODULE_STACK];
     double scratch[2][PURE_LIST_RUN];
@@ -1064,6 +1192,16 @@ static bool run_lanes(const PureModule *m, const double *values, PureBlock block
         uint32_t arg = op->arg;
         Lanes *top = sp ? &stack[sp - 1] : stack;
         double *out = store[sp ? sp - 1 : 0]; /* where the top's new lanes go */
+        /* A group's pass: every group at once, so its key is not needed and its test holds. */
+        if (shape && pc == shape->key.start) {
+            stack[sp++] = (Lanes){NULL, 0, false, 0};
+            pc = shape->key.start + shape->key.length - 1;
+            continue;
+        }
+        if (shape && pc == shape->test) {
+            stack[--sp - 1] = (Lanes){NULL, 1, false, 0};
+            continue;
+        }
         switch (op->op) {
         case PURE_OP_CONST: stack[sp++] = (Lanes){NULL, op->value, false, 0}; continue;
         case PURE_OP_LOAD: stack[sp++] = (Lanes){NULL, values[arg], false, 0}; continue;
@@ -1253,7 +1391,7 @@ static PureResult fold_runs(const PureModule *m, double *values, PureBlock inner
     memo.known = 0;
     for (bool more = pure_list_run(list, 0, &items); more && !fold->done; more = pure_list_next(list, &items)) {
         Lanes out[2];
-        if (!run_lanes(m, values, inner, per, &items, out, store, &memo)) {
+        if (!run_lanes(m, values, inner, per, &items, out, store, &memo, NULL)) {
             PureResult status = fold_rows(m, values, inner, per, items.start, items.start + items.count, fold);
             if (status != PURE_OK) return status;
             continue;
@@ -1279,6 +1417,225 @@ static PureResult fold_runs(const PureModule *m, double *values, PureBlock inner
     return PURE_OK;
 }
 
+/* ---- Passes: a shaped fold in a column answered for every row --------
+ * A fold in a column runs once per row, over its whole list each time. A
+ * shaped one (PureShape) is answered from one pass over the list instead,
+ * kept by the app for the list's version: a group pass folds the items of
+ * every group at once (the fold's test with its comparison taken as true,
+ * the item's field naming its group), a rank pass orders the field's values.
+ * A row then runs the key alone and looks it up. The pass is made at the
+ * second row that asks for the same list, so a fold asked once scans once.
+ *
+ * It answers exactly as the fold would: items join a group in order (a sum
+ * adds as it would), a key that faults leaves the row to run the fold as
+ * written (which faults, or not over no item, as it would), and an item
+ * that would fault in the pass leaves every row to run it as written. */
+typedef struct { double key; Fold fold; bool used; } Group;
+struct PurePass {
+    const PureModule *module;
+    uint32_t pc;             /* the fold's */
+    uint64_t list, version;  /* its list as it was */
+    unsigned asked;          /* rows that asked before it was made */
+    bool made, failed;
+    Group *groups;           /* a group pass: open addressing by key */
+    size_t capacity, count;
+    uint64_t *ordered;       /* a rank pass: the field's values in order (order_bits) */
+    size_t items;
+};
+enum { PASSES = 16 };
+
+static void free_pass(struct PurePass *pass) {
+    free(pass->groups);
+    free(pass->ordered);
+}
+
+unsigned pure_app_passes(const PureApp *app) {
+    unsigned made = 0;
+    for (unsigned i = 0; i < app->pass_count; ++i) made += app->passes[i].made;
+    return made;
+}
+
+static void drop_passes(PureApp *app) {
+    for (unsigned i = 0; i < app->pass_count; ++i) free_pass(&app->passes[i]);
+    free(app->passes);
+    app->passes = NULL;
+    app->pass_count = 0;
+}
+
+/* The app's pass for a fold over a list as it is: one kept, or a new one in
+ * place of this fold's older ones (or of the oldest). NULL without room. */
+static struct PurePass *pass_for(PureApp *app, const PureModule *m, uint32_t pc, const PureList *list) {
+    uint64_t id, version;
+    pure_list_stamp(list, &id, &version);
+    unsigned kept = 0;
+    struct PurePass *found = NULL;
+    for (unsigned i = 0; i < app->pass_count; ++i) {
+        struct PurePass *pass = &app->passes[i];
+        if (pass->module == m && pass->pc == pc && pass->list == id && pass->version == version) found = pass;
+        else if (pass->module == m && pass->pc == pc) free_pass(pass); /* the list has moved on */
+        else app->passes[kept++] = *pass;
+        if (found == pass) app->passes[kept++] = *pass, found = &app->passes[kept - 1];
+    }
+    app->pass_count = kept;
+    if (found) return found;
+    if (!app->passes && !(app->passes = calloc(PASSES, sizeof(*app->passes)))) return NULL;
+    if (app->pass_count == PASSES) {
+        free_pass(&app->passes[0]);
+        memmove(app->passes, app->passes + 1, (PASSES - 1) * sizeof(*app->passes));
+        --app->pass_count;
+    }
+    struct PurePass *pass = &app->passes[app->pass_count++];
+    *pass = (struct PurePass){.module = m, .pc = pc, .list = id, .version = version};
+    return pass;
+}
+
+/* A key's bits as they order: -0 as 0, negative numbers below positive ones. */
+static uint64_t order_bits(double value) {
+    uint64_t bits;
+    value = value == 0 ? 0 : value;
+    memcpy(&bits, &value, sizeof(bits));
+    return bits >> 63 ? ~bits : bits | (uint64_t)1 << 63;
+}
+
+static size_t group_slot(uint64_t bits, size_t capacity) {
+    bits ^= bits >> 33, bits *= 0xff51afd7ed558ccdu, bits ^= bits >> 33; /* a finalizer's mixing */
+    return (size_t)bits & (capacity - 1);
+}
+
+/* key's group, added when absent (with add); NULL when absent, or without room. */
+static Group *group_of(struct PurePass *pass, double key, unsigned how, bool add) {
+    key = key == 0 ? 0 : key;
+    if (pass->capacity) {
+        for (size_t at = group_slot(order_bits(key), pass->capacity);; at = (at + 1) & (pass->capacity - 1)) {
+            if (!pass->groups[at].used) break;
+            if (pass->groups[at].key == key) return &pass->groups[at];
+        }
+    }
+    if (!add) return NULL;
+    if (2 * (pass->count + 1) > pass->capacity) { /* grown to keep half the slots free */
+        size_t capacity = pass->capacity ? 2 * pass->capacity : 64;
+        Group *groups = calloc(capacity, sizeof(*groups));
+        if (!groups) return NULL;
+        for (size_t i = 0; i < pass->capacity; ++i) {
+            if (!pass->groups[i].used) continue;
+            size_t at = group_slot(order_bits(pass->groups[i].key), capacity);
+            while (groups[at].used) at = (at + 1) & (capacity - 1);
+            groups[at] = pass->groups[i];
+        }
+        free(pass->groups);
+        pass->groups = groups;
+        pass->capacity = capacity;
+    }
+    size_t at = group_slot(order_bits(key), pass->capacity);
+    while (pass->groups[at].used) at = (at + 1) & (pass->capacity - 1);
+    pass->groups[at] = (Group){key, {how, 0, false, false}, true};
+    ++pass->count;
+    return &pass->groups[at];
+}
+
+/* Every group's fold in one pass over the list, a leaf at a time. */
+static bool group_pass(const PureModule *m, double *values, PureBlock inner, const PureShape *shape, unsigned how,
+                       const PureList *list, Event *per, struct PurePass *pass) {
+    double store[PURE_MODULE_STACK][PURE_LIST_RUN], scratch[2][PURE_LIST_RUN];
+    bool one = how == PURE_FOLD_COUNT || how == PURE_FOLD_FIRST;
+    PureRun items;
+    Memo memo;
+    memo.known = 0;
+    for (bool more = pure_list_run(list, 0, &items); more; more = pure_list_next(list, &items)) {
+        Lanes out[2];
+        if (!run_lanes(m, values, inner, per, &items, out, store, &memo, shape)) return false;
+        const Lanes *counts = &out[one ? 0 : 1];
+        if (uniform(counts) && counts->value == 0) continue;
+        const double *c = lanes_of(counts, scratch[0], &items), *v = lanes_of(&out[0], scratch[1], &items);
+        const double *keys = items.column[shape->field].numbers;
+        for (unsigned i = 0; i < items.count; ++i) {
+            if (c[i] == 0) continue;
+            Group *group = group_of(pass, keys[i], how, true);
+            if (!group) return false;
+            if (!group->fold.done) fold_in(&group->fold, items.start + i, v[i], c[i]);
+        }
+    }
+    return true;
+}
+
+/* The field's values in order: a radix sort of their order_bits. */
+static bool rank_pass(const PureShape *shape, const PureList *list, struct PurePass *pass) {
+    size_t count = pure_list_count(list), at = 0;
+    uint64_t *keys = malloc((count ? count : 1) * sizeof(*keys)), *spare = malloc((count ? count : 1) * sizeof(*spare));
+    if (!keys || !spare) {
+        free(keys);
+        free(spare);
+        return false;
+    }
+    PureRun items;
+    for (bool more = pure_list_run(list, 0, &items); more; more = pure_list_next(list, &items))
+        for (unsigned i = 0; i < items.count; ++i) keys[at++] = order_bits(items.column[shape->field].numbers[i]);
+    for (unsigned shift = 0; shift < 64; shift += 8) {
+        size_t start[257] = {0};
+        for (size_t i = 0; i < count; ++i) ++start[(keys[i] >> shift & 0xFF) + 1];
+        if (start[(keys[0] >> shift & 0xFF) + 1] == count) continue; /* one digit for all: in order already */
+        for (unsigned d = 0; d < 256; ++d) start[d + 1] += start[d];
+        for (size_t i = 0; i < count; ++i) spare[start[keys[i] >> shift & 0xFF]++] = keys[i];
+        uint64_t *swap = keys;
+        keys = spare, spare = swap;
+    }
+    free(spare);
+    pass->ordered = keys;
+    pass->items = count;
+    return true;
+}
+
+/* How many of the ordered values relate to key: the first not below key
+ * (or above it), counted from the start or the end. */
+static size_t ranked(const struct PurePass *pass, uint8_t relation, double key) {
+    uint64_t bits = order_bits(key);
+    bool above = relation == PURE_OP_GT || relation == PURE_OP_LE; /* the bound past key's equals */
+    size_t low = 0, high = pass->items;
+    while (low < high) {
+        size_t middle = low + (high - low) / 2;
+        if (above ? pass->ordered[middle] <= bits : pass->ordered[middle] < bits) low = middle + 1;
+        else high = middle;
+    }
+    return relation == PURE_OP_GT || relation == PURE_OP_GE ? pass->items - low : low;
+}
+
+/* A shaped fold from its list's pass: false leaves it to run as written. */
+static bool fold_shaped(const PureModule *m, double *values, const PureOp *op, const PureList *list, Event *per,
+                        Fold *fold) {
+    PureApp *app = per->app;
+    const PureShape *shape = &m->shape[op->shape - 1];
+    struct PurePass *pass = pass_for(app, m, (uint32_t)(op - m->code), list);
+    if (!pass || pass->failed || (!pass->made && ++pass->asked < 2)) return false;
+    Slot key;
+    PureFault fault = app->fault;
+    uint32_t fault_pc = app->fault_pc;
+    size_t mark = per->arena->used;
+    PureResult status = run(m, values, shape->key, per, &key);
+    per->arena->used = mark;
+    app->fault = fault, app->fault_pc = fault_pc;
+    if (status != PURE_OK) return false;
+    if (!pass->made) {
+        PureBlock inner = {(uint32_t)(op - m->code) + 1, op->arg >> 16};
+        pass->made = shape->form == PURE_SHAPE_GROUP ? group_pass(m, values, inner, shape, fold->how, list, per, pass)
+                                                     : rank_pass(shape, list, pass);
+        if (!pass->made) {
+            free_pass(pass);
+            pass->groups = NULL, pass->ordered = NULL;
+            pass->failed = true;
+            return false;
+        }
+    }
+    if (shape->form == PURE_SHAPE_RANK) {
+        size_t count = ranked(pass, shape->relation, key.num);
+        fold->total = (double)count;
+        fold->any = count > 0;
+        return true;
+    }
+    const Group *group = group_of(pass, key.num, fold->how, false);
+    if (group) *fold = group->fold;
+    return true;
+}
+
 static PureResult run(const PureModule *m, double *values, PureBlock block, Event *ev, Slot *result) {
     Slot stack[PURE_MODULE_STACK];
     size_t sp = 0;
@@ -1538,9 +1895,11 @@ static PureResult run(const PureModule *m, double *values, PureBlock block, Even
             Fold fold = {arg & 0xFF, 0, false, false};
             Event per = *ev;
             per.outer = ev->row;
-            status = op->plan ? fold_runs(m, values, inner, list, &per, &fold)
-                              : fold_rows(m, values, inner, &per, 0, pure_list_count(list), &fold);
-            if (status != PURE_OK) return status;
+            if (!op->shape || !fold_shaped(m, values, op, list, &per, &fold)) {
+                status = op->plan ? fold_runs(m, values, inner, list, &per, &fold)
+                                  : fold_rows(m, values, inner, &per, 0, pure_list_count(list), &fold);
+                if (status != PURE_OK) return status;
+            }
             if (!fold.any && fold.how >= PURE_FOLD_MAX) { kind = PURE_FAULT_NOT_FOUND; goto failed; }
             if (!isfinite(fold.total)) { kind = PURE_FAULT_NON_FINITE; goto failed; }
             stack[sp].num = fold.total;
@@ -1818,6 +2177,7 @@ void pure_app_stop(PureApp *app) {
     pure_undo_free(&app->undo);
     for (unsigned i = 0; i < app->row_tables; ++i) free(app->rows[i].items);
     free(app->rows);
+    drop_passes(app);
     free(app->arena.bytes);
     free(app->view_arena.bytes);
     memset(app, 0, sizeof(*app));
@@ -2173,7 +2533,7 @@ static struct PureRows *filtered(PureApp *app, uint32_t table) {
         memo.known = 0;
         for (bool more = pure_list_run(list, 0, &run); more; more = pure_list_next(list, &run)) {
             Lanes keep;
-            if (run_lanes(m, app->values, m->view_code[table], &ev, &run, &keep, store, &memo)) {
+            if (run_lanes(m, app->values, m->view_code[table], &ev, &run, &keep, store, &memo, NULL)) {
                 const double *kept = lanes_of(&keep, scratch, &run);
                 for (unsigned i = 0; i < run.count; ++i)
                     if (kept[i] != 0 && !keep_row(rows, run.start + i)) return NULL;
@@ -2297,6 +2657,7 @@ PureSnapshot *pure_app_snapshot(PureApp *app) {
 }
 
 void pure_app_read(PureApp *reader, const PureSnapshot *snapshot) {
+    if (reader->module != snapshot->module) drop_passes(reader);
     reader->module = snapshot->module;
     reader->reading = true;
     reader->changes = snapshot->changes; /* what the row cache is fresh for */
@@ -2514,6 +2875,7 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
     memcpy(app->list_drafts, list_drafts, sizeof(list_drafts));
     memcpy(app->saved, saved, sizeof(saved));
     if (old) remap_outputs(app, old, module, true);
+    drop_passes(app); /* kept by instruction of the module they were made for */
     app->module = module;
     ++app->changes;
     return PURE_OK;
diff --git a/core/module.h b/core/module.h
index 9a491c9..2da5c6b 100644
--- a/core/module.h
+++ b/core/module.h
@@ -84,9 +84,11 @@ typedef enum {
     PURE_FAULT_LIMIT, PURE_FAULT_NOT_FOUND, PURE_FAULT_ENCODING, PURE_FAULT_EXISTS
 } PureFault;
 
-/* plan is the runtime's, set at load: a fold whose block has a form for
- * every instruction that works on a leaf of its list at once runs so. */
-typedef struct { double value; uint32_t arg; uint8_t op, plan; } PureOp;
+/* plan and shape are the runtime's, set at load: a fold whose block has a
+ * form for every instruction that works on a leaf of its list at once runs
+ * so (plan), and one in a column may be answered for every row from one
+ * pass over its list (shape: its PureShape + 1). */
+typedef struct { double value; uint32_t arg; uint8_t op, plan, shape; } PureOp;
 typedef struct { uint32_t start, length; } PureBlock;
 typedef struct { uint32_t event; PureBlock code; } PureHandler;
 typedef struct { char name[PURE_MODULE_NAME]; uint8_t type; uint32_t offset; } PureField;
@@ -101,6 +103,15 @@ typedef struct {
 /* A record type: fields numbered from first, each a number or text. */
 typedef struct { uint16_t first, count; } PureRecord;
 
+/* A fold in a column whose test compares a field of its item with a key the
+ * row gives (key, a block reading no item; test, the comparison), the rest
+ * of it reading the item alone: a group (the field equals the key: one pass
+ * folds every group) or a rank (a count of items whose field orders against
+ * the key: one sort answers every key). relation is the comparison's
+ * opcode with the item on its left. */
+enum { PURE_SHAPE_GROUP = 1, PURE_SHAPE_RANK = 2, PURE_MODULE_SHAPES = 64 };
+typedef struct { uint8_t form, relation; uint16_t field; uint32_t test; PureBlock key; } PureShape;
+
 typedef struct {
     uint16_t states, inputs, views, handlers, fields, collections, constants, records, record_fields;
     uint32_t code_count;
@@ -123,6 +134,8 @@ typedef struct {
     uint16_t view_list[PURE_MODULE_VIEWS];  /* the list of a table or column */
     PureBlock view_code[PURE_MODULE_VIEWS];
     uint8_t view_plan[PURE_MODULE_VIEWS]; /* a table's filter runs a leaf of its list at once (as a fold may) */
+    PureShape shape[PURE_MODULE_SHAPES];
+    uint8_t shapes;
     PureHandler handler[PURE_MODULE_HANDLERS];
     /* The code and the text constants, sized to the module in one block
      * (storage) that load takes and pure_module_release gives back. */
@@ -255,6 +268,9 @@ typedef struct {
     bool links_changed; /* since the last save */
     bool duplicate;     /* the last delivery was heard before and not applied */
     struct { const char *source; uint64_t seq; bool pending; } delivering;
+    /* Passes over lists that answer shaped folds, by the list's version. */
+    struct PurePass *passes;
+    unsigned pass_count;
     struct PureSnapshot *snapshot; /* the newest snapshot made, or NULL */
     PureRetired retired;           /* list items let go since it was made */
     bool reading;                  /* a reader: its states are a snapshot's */
@@ -304,6 +320,8 @@ PureResult pure_app_view_text(PureApp *, uint32_t view, bool (*visit)(void *, Pu
 /* The rows of a table view (its list's items, or those its filter keeps),
  * or 0 for another view. A host evaluates only the cells it shows. */
 size_t pure_app_rows(PureApp *, uint32_t table);
+/* How many passes the app keeps, made to answer shaped folds in columns. */
+unsigned pure_app_passes(const PureApp *);
 /* The item a row of a table shows, and the row that shows an item (false:
  * its filter leaves the item out). Without a filter both are the identity. */
 bool pure_app_row_item(PureApp *, uint32_t table, size_t row, size_t *item);
diff --git a/core/test-module.c b/core/test-module.c
index b222597..6c76510 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -2179,7 +2179,9 @@ static void filters(void) {
  * a handler storing them), filters and lists: a module whose folds run a
  * leaf at a time answers bit for bit as the same module run item by item,
  * faults included (their kind and instruction). */
-typedef struct { O code[8192]; size_t count; uint32_t seed; int fold_list, outer; } Gen;
+/* tree: no PICK or SLIDE (a shaped fold's block is a tree). A fold_list of
+ * -1 makes code that reads no fold's item. */
+typedef struct { O code[8192]; size_t count; uint32_t seed; int fold_list, outer; bool tree; } Gen;
 enum { LANE_K = 0, LANE_L = 4, LANE_M = 5 };
 
 static void put_op(Gen *g, uint8_t op, uint32_t arg, double value) {
@@ -2213,11 +2215,13 @@ static void gen_num(Gen *g, int depth) {
                                      PURE_OP_GT, PURE_OP_GE, PURE_OP_EQ, PURE_OP_NE, PURE_OP_AND, PURE_OP_OR,
                                      PURE_OP_MIN, PURE_OP_MAX, PURE_OP_MOD};
     static const uint8_t unary[] = {PURE_OP_NEG, PURE_OP_NOT, PURE_OP_FLOOR};
-    switch (roll(g, depth <= 0 ? 4 : 12)) {
+    unsigned choice = roll(g, depth <= 0 ? 4 : 12);
+    if (g->tree && (choice == 9 || choice == 10)) choice = 11;
+    switch (choice) {
     case 0: put_op(g, C, 0, pool[roll(g, LEN(pool))]); return;
     case 1: put_op(g, L, LANE_K, 0); return;
     case 2: case 3: { /* a field of an item */
-        int list = depth <= 0 || roll(g, 3) ? g->fold_list : roll(g, 2) ? LANE_L : LANE_M;
+        int list = g->fold_list >= 0 && (depth <= 0 || roll(g, 3)) ? g->fold_list : roll(g, 2) ? LANE_L : LANE_M;
         gen_index(g, list, depth - 1);
         put_op(g, PURE_OP_ITEM_GET, (uint32_t)list | (uint32_t)roll(g, list == LANE_L ? 3 : 2) << 16, 0);
         return;
@@ -2236,6 +2240,66 @@ static void gen_num(Gen *g, int depth) {
     }
 }
 
+/* What reads only the fold's list, its item and constants (a shaped fold's own parts). */
+static void gen_own(Gen *g, int depth) {
+    static const double pool[] = {0, -0.0, 1, -1, 2, 0.5, 3, 5, 1e308};
+    static const uint8_t binary[] = {PURE_OP_ADD, PURE_OP_SUB, PURE_OP_MUL, PURE_OP_DIV, PURE_OP_LT, PURE_OP_GT,
+                                     PURE_OP_EQ, PURE_OP_NE, PURE_OP_AND, PURE_OP_OR, PURE_OP_MIN, PURE_OP_MOD};
+    int list = g->fold_list;
+    switch (roll(g, depth <= 0 ? 3 : 7)) {
+    case 0: put_op(g, C, 0, pool[roll(g, LEN(pool))]); return;
+    case 1: case 2:
+        if (roll(g, 4)) put_op(g, PURE_OP_ROW, (uint32_t)list, 0);
+        else put_op(g, roll(g, 2) ? PURE_OP_INDEX_FIRST : PURE_OP_INDEX_LAST, (uint32_t)list, 0);
+        put_op(g, PURE_OP_ITEM_GET, (uint32_t)list | (uint32_t)roll(g, list == LANE_L ? 3 : 2) << 16, 0);
+        return;
+    case 3: put_op(g, PURE_OP_LIST_COUNT, (uint32_t)list, 0); return;
+    case 4: gen_own(g, depth - 1); put_op(g, roll(g, 2) ? NEG : PURE_OP_NOT, 0, 0); return;
+    default: gen_own(g, depth - 1); gen_own(g, depth - 1); put_op(g, binary[roll(g, LEN(binary))], 0, 0); return;
+    }
+}
+
+/* A fold in a column of L with a shape's test: the item's field against a
+ * key the row gives, ANDed with the item's own tests, and an own value. */
+static void gen_shaped(Gen *g, int list) {
+    static const uint8_t relations[] = {PURE_OP_EQ, PURE_OP_EQ, PURE_OP_LT, PURE_OP_LE, PURE_OP_GT, PURE_OP_GE};
+    unsigned how = roll(g, 5) < 2 ? PURE_FOLD_COUNT : roll(g, 5), before = roll(g, 3) ? 0 : 1 + roll(g, 2),
+             after = roll(g, 3) ? 0 : 1;
+    size_t at = g->count;
+    put_op(g, PURE_OP_FOLD, 0, 0);
+    int fold_list = g->fold_list, fold_outer = g->outer;
+    g->fold_list = list, g->outer = LANE_L, g->tree = true;
+    if (how != PURE_FOLD_COUNT && how != PURE_FOLD_FIRST) gen_own(g, 2);
+    for (unsigned k = 0; k < before; ++k) {
+        gen_own(g, 2);
+        if (k) put_op(g, PURE_OP_AND, 0, 0);
+    }
+    bool field_first = roll(g, 2);
+    for (unsigned side = 0; side < 2; ++side) {
+        if (side == !field_first) { /* the item's field */
+            put_op(g, PURE_OP_ROW, (uint32_t)list, 0);
+            put_op(g, PURE_OP_ITEM_GET, (uint32_t)list | (uint32_t)roll(g, list == LANE_L ? 3 : 2) << 16, 0);
+        } else { /* the key: the column's row's field, mostly */
+            g->fold_list = -1;
+            if (roll(g, 5)) {
+                put_op(g, PURE_OP_OUTER_ROW, LANE_L, 0);
+                put_op(g, PURE_OP_ITEM_GET, LANE_L | (uint32_t)roll(g, 3) << 16, 0);
+            } else
+                gen_num(g, 2);
+            g->fold_list = list;
+        }
+    }
+    put_op(g, relations[roll(g, LEN(relations))], 0, 0);
+    if (before) put_op(g, PURE_OP_AND, 0, 0);
+    for (unsigned k = 0; k < after; ++k) {
+        gen_own(g, 2);
+        put_op(g, PURE_OP_AND, 0, 0);
+    }
+    g->fold_list = fold_list, g->outer = fold_outer, g->tree = false;
+    g->code[at].arg = FOLDING(how, list, g->count - at - 1);
+    if (how == PURE_FOLD_FIRST) put_op(g, PURE_OP_INDEX_NUM, 0, 0);
+}
+
 /* A fold over list leaving a number (first's index as one), in a column of L (outer) or not. */
 static void gen_fold(Gen *g, int list, int outer) {
     unsigned how = roll(g, 5);
@@ -2264,7 +2328,7 @@ static void lane_folds(void) {
     enum { VIEWS = 6, COLUMNS = 3, HANDLED = 3 };
     static Gen g;
     uint32_t seed = 0x1a2b3c4du;
-    unsigned planned = 0, folds = 0, faults = 0, cells = 0;
+    unsigned planned = 0, folds = 0, faults = 0, cells = 0, shaped[3] = {0}, passes = 0;
     for (unsigned round = 0; round < 300; ++round) {
         g = (Gen){.seed = next_random(&seed), .fold_list = -1, .outer = -1};
         V views[VIEWS];
@@ -2284,7 +2348,8 @@ static void lane_folds(void) {
         for (unsigned c = 0; c < COLUMNS; ++c) {
             snprintf(names[VIEWS + 1 + c], sizeof(names[0]), "c%u", c);
             tables[1 + c] = (W){names[VIEWS + 1 + c], 0, VIEWS + 1, (uint32_t)g.count, 0};
-            gen_fold(&g, roll(&g, 3) ? LANE_L : LANE_M, LANE_L);
+            if (roll(&g, 3)) gen_shaped(&g, roll(&g, 3) ? LANE_L : LANE_M);
+            else gen_fold(&g, roll(&g, 3) ? LANE_L : LANE_M, LANE_L);
             tables[1 + c].length = (uint32_t)(g.count - tables[1 + c].start);
         }
         H handler = {1, (uint32_t)g.count, 0};
@@ -2302,7 +2367,9 @@ static void lane_folds(void) {
         for (uint32_t i = 0; i < module_b.code_count; ++i) {
             folds += module_a.code[i].op == PURE_OP_FOLD;
             planned += module_a.code[i].plan;
-            module_b.code[i].plan = 0;
+            shaped[module_a.code[i].shape ? module_a.shape[module_a.code[i].shape - 1].form : 0] +=
+                module_a.code[i].op == PURE_OP_FOLD;
+            module_b.code[i].plan = module_b.code[i].shape = 0;
         }
         planned += module_a.view_plan[VIEWS];
         memset(module_b.view_plan, 0, sizeof(module_b.view_plan));
@@ -2341,6 +2408,7 @@ static void lane_folds(void) {
                 }
             }
         }
+        passes += pure_app_passes(&a);
         PureResult went_a = pure_app_input(&a, 0, PURE_EDGE_DOWN), went_b = pure_app_input(&b, 0, PURE_EDGE_DOWN);
         if (went_a != went_b || a.fault != b.fault || a.fault_pc != b.fault_pc || memcmp(a.values, b.values, 4 * sizeof(double))) {
             fprintf(stderr, "lane round %u handler: %d %d %u / %d %d %u\n", round, went_a, a.fault, a.fault_pc, went_b, b.fault,
@@ -2353,8 +2421,10 @@ static void lane_folds(void) {
         pure_module_release(&module_a);
         pure_module_release(&module_b);
     }
-    CHECK(planned > folds / 2 && faults > 100 && cells > 1000);
-    printf("{\"lane_folds\":%u,\"planned\":%u,\"faults_alike\":%u,\"cells\":%u}\n", folds, planned, faults, cells);
+    CHECK(planned > folds / 2 && faults > 100 && cells > 1000 && shaped[PURE_SHAPE_GROUP] > 150 &&
+          shaped[PURE_SHAPE_RANK] > 100 && passes > 60);
+    printf("{\"lane_folds\":%u,\"planned\":%u,\"groups\":%u,\"ranks\":%u,\"passes\":%u,\"faults_alike\":%u,"
+           "\"cells\":%u}\n", folds, planned, shaped[PURE_SHAPE_GROUP], shaped[PURE_SHAPE_RANK], passes, faults, cells);
 }
 
 /* Damaged fold and filter modules: whatever loads runs safely. */