sort into a new table.

/// Tasks are (column, row-chunk) pairs: pass 1 measures bytes per task, pass 2 copies.
fn materialize_mt(t: &Table, perm: &[u32], pool: &Pool, parts: usize) -> Table {
    let n = perm.len();
    let nc = t.cols.len();
    let ranges = chunk_ranges(n, parts);
    let sizes: Vec<usize> = pool.map(nc * parts, |task| {
        let (c, p) = (task / parts, task % parts);
        let (a, b) = ranges[p];
        gathered_bytes(&t.cols[c], perm, a, b)
    });
    let stride = parts + 1;
    let mut bases = vec![0usize; nc * stride];
    for c in 0..nc {
        let mut s = 0usize;
        for p in 0..parts {
            bases[c * stride + p] = s;
            s += sizes[c * parts + p];
        }
        bases[c * stride + parts] = s;
    }
    let mut outs: Vec<(Vec<u32>, Vec<u8>)> =
        (0..nc).map(|c| (Vec::with_capacity(n + 1), Vec::with_capacity(bases[c * stride + parts] + PAD))).collect();
    let ptrs: Vec<(SyncPtr<u32>, SyncPtr<u8>)> =
        outs.iter_mut().map(|(o, a)| (SyncPtr(o.as_mut_ptr()), SyncPtr(a.as_mut_ptr()))).collect();
    pool.run(nc * parts, |task| {
        let (c, p) = (task / parts, task % parts);
        let (a, b) = ranges[p];
        let (op, ap) = ptrs[c];
        let base = bases[c * stride + p];
        let end = bases[c * stride + p + 1];
        unsafe { gather_range(&t.cols[c], perm, a, b, base, end, ap.get(), op.get()) };
    });
    let cols = outs
        .into_iter()
        .enumerate()
        .map(|(c, (mut offs, mut arena))| {
            let total = bases[c * stride + parts];
            unsafe {
                *offs.as_mut_ptr() = 0;
                offs.set_len(n + 1);
                std::ptr::write_bytes(arena.as_mut_ptr().add(total), 0, PAD);
                arena.set_len(total + PAD);
            }
            Column { offs, arena }
        })
        .collect();
    Table { headers: t.headers.clone(), rows: n, cols }
}

fn materializeMulti(pool: *Pool, t: *const Table, perm: []const u32, m: *Mat, scratch: [][MAX_COLS]usize, ends: [][MAX_COLS]usize) void {
    const nk = scratch.len;
    var j: MatJob = .{ .t = t, .perm = perm, .m = m, .nk = nk, .bytes = scratch, .ends = ends };
    pool.runChunks(&j, MatJob.sizeJob, t.ncols * nk);
    var run: [MAX_COLS]usize = @splat(0);
    for (0..nk) |k| {
        for (0..t.ncols) |c| {
            const b = j.bytes[k][c];
            j.bytes[k][c] = run[c];
            run[c] += b;
            j.ends[k][c] = run[c];
        }
    }
    for (0..t.ncols) |c| m.cols[c].offsets[0] = 0;
    pool.runChunks(&j, MatJob.copyJob, (t.ncols + 1) * nk); // the extra column of items gathers the typed arrays
    for (0..t.ncols) |c| m.cols[c].len = run[c];
}

func materialize(t *Table, perm []uint32, p *pool) *Table {
	n, ncols := t.Rows, len(t.Cols)
	out := &Table{Names: t.Names, Rows: n, Cols: make([]Column, ncols)}
	p.run(ncols, ncols, func(c, _, _ int) {
		out.Cols[c] = Column{Arena: make([]byte, int(t.Cols[c].Offs[n])+arenaSlack), Offs: make([]uint32, n+1)}
	})
	C := p.chunks()
	sizes := make([]int, C+1)
	for c := 0; c < ncols; c++ {
		sa, so := t.Cols[c].Arena, t.Cols[c].Offs
		da, do := out.Cols[c].Arena, out.Cols[c].Offs
		if C == 1 {
			sizes[1] = int(so[n])
		} else {
			p.run(C, n, func(ch, lo, hi int) {
				sum := 0
				for _, src := range perm[lo:hi] {
					sum += int(so[src+1] - so[src])
				}
				sizes[ch+1] = sum
			})
			for ch := 0; ch < C; ch++ {
				sizes[ch+1] += sizes[ch]
			}
		}
		p.run(C, n, func(ch, lo, hi int) {
			w, end := sizes[ch], sizes[ch+1]
			for r := lo; r < hi; r++ {
				q := perm[r]
				s, e := int(so[q]), int(so[q+1])
				l := e - s
				if l <= 8 && w+8 <= end {
					binary.LittleEndian.PutUint64(da[w:], binary.LittleEndian.Uint64(sa[s:]))
				} else if l > 8 && l <= 16 {
					binary.LittleEndian.PutUint64(da[w:], binary.LittleEndian.Uint64(sa[s:]))
					binary.LittleEndian.PutUint64(da[w+l-8:], binary.LittleEndian.Uint64(sa[s+l-8:]))
				} else {
					copy(da[w:], sa[s:e])
				}
				w += l
				do[r+1] = uint32(w)
			}
		})
	}
	return out
}

// ───────────────────────────── materialise the sorted table ─────────────────────────────
/// Gathers every column of `t` by `perm` into fresh arenas + offsets (same total bytes).
func materialize(_ t: Table, _ perm: UnsafePointer<UInt32>, pool: Pool?, parts: Int) -> [Column] {
    let n = t.rows
    var out: [Column] = []
    for c in t.cols {
        let off = alloc(n + 1, UInt32.self)
        off[0] = 0
        out.append(Column(arena: allocBytes(c.bytes + 64), offsets: off, bytes: c.bytes))
    }
    let ncols = t.cols.count
    @inline(__always) func copyCell(_ dst: UnsafeMutableRawPointer, _ src: UnsafeRawPointer, _ len: Int) {
        dst.storeBytes(of: src.loadUnaligned(as: V16.self), as: V16.self)
        if len > 16 { var k = 16; while k < len { (dst + k).storeBytes(of: (src + k).loadUnaligned(as: V16.self), as: V16.self); k += 16 } }
    }
    guard let pool = pool, parts > 1 else {
        for c in 0..<ncols {
            let src = t.cols[c], dst = out[c]
            var cursor = 0
            var i = 0
            while i < n {
                let r = Int(perm[i])
                let s = Int(src.offsets[r]), len = Int(src.offsets[r + 1]) - s
                copyCell(UnsafeMutableRawPointer(dst.arena + cursor), UnsafeRawPointer(src.arena + s), len)
                cursor += len
                dst.offsets[i + 1] = UInt32(truncatingIfNeeded: cursor)
                i += 1
            }
        }
        return out
    }
    let np = max(1, min(parts, n / 4096))
    let chunk = (n + np - 1) / np
    let partBytes = alloc(ncols * (np + 1), Int.self)   // [c * (np+1) + p]
    // 1. lengths (into the new offsets) and bytes per (column, part)
    pool.run(ncols * np) { task in
        let c = task % ncols, p = task / ncols
        let src = t.cols[c], dst = out[c]
        var total = 0
        var i = p * chunk
        let hi = min(n, i + chunk)
        while i < hi {
            let r = Int(perm[i])
            let len = Int(src.offsets[r + 1]) - Int(src.offsets[r])
            dst.offsets[i + 1] = UInt32(truncatingIfNeeded: len)
            total += len
            i += 1
        }
        partBytes[c * (np + 1) + p] = total
    }
    for c in 0..<ncols {
        var running = 0
        for p in 0..<np { let b = partBytes[c * (np + 1) + p]; partBytes[c * (np + 1) + p] = running; running += b }
        partBytes[c * (np + 1) + np] = running
    }
    // 2. copy with known destinations. The 16-byte vector copy may spill past a cell; inside a
    //    part that is overwritten by the next cell, but the last ≤16 bytes of a part belong to
    //    another thread, so cells that close to the part end are copied exactly.
    pool.run(ncols * np) { task in
        let c = task % ncols, p = task / ncols
        let src = t.cols[c], dst = out[c]
        var cursor = partBytes[c * (np + 1) + p]
        let safeEnd = partBytes[c * (np + 1) + p + 1] - 16
        var i = p * chunk
        let hi = min(n, i + chunk)
        while i < hi {
            let r = Int(perm[i])
            let len = Int(dst.offsets[i + 1])
            let d = UnsafeMutableRawPointer(dst.arena + cursor), sp = UnsafeRawPointer(src.arena + Int(src.offsets[r]))
            if cursor + len <= safeEnd { copyCell(d, sp, len) } else { memcpy(d, sp, len) }
            cursor += len
            dst.offsets[i + 1] = UInt32(truncatingIfNeeded: cursor)
            i += 1
        }
    }
    free(partBytes)
    return out
}

timed("sort_mat", lambda: df.sort("amount"))

timed("sort_mat", lambda: con.execute("CREATE OR REPLACE TABLE s AS SELECT * FROM t ORDER BY amount"))

timed("sort_mat", lambda: df.sort_values("amount"))

timed("sort_mat", lambda: sink(df.orderBy("amount")))

 194  l.t = csv::parse(raw);
 337  const double f5 = timeIt([&] { Bitmap b = compareDoubles(amt.data(), n, Cmp::Gt, 500, &pool); sink = b.words[0]; }, 7);
 372  const double s4 = timeIt([&] { sortIndexByDouble(amt.data(), n, perm, &pool); sink = perm[0]; }, 3);
 454  const double a5 = G <= 16 ? timeIt([&] { groupSumAnyOrder(amtP, idP, n, G, &pool, pairTotal); sink = static_cast<size_t>(pairTotal[0]); }, 9) : 0;

// sort.h — stable O(n) sorting of row indices by a double key (parallel radix).
// JS numeric sort semantics: NaN sorts as -Infinity; ties keep input order.

// Fills `idx` with the stable ascending order of keys[0..n). pool may be null.
void sortIndexByDouble(const double* keys, size_t n, std::vector<uint32_t>& idx, Pool* pool);