filter amount > 500, count.

fn filterMulti(pool: *Pool, amount: []const f64, out: []u32, counts: []usize, bases: []usize) usize {
    const nk = counts.len;
    var j: FilterJob = .{ .amount = amount, .out = out, .counts = counts, .bases = bases };
    pool.runChunks(&j, FilterJob.countJob, nk);
    var sum: usize = 0;
    for (0..nk) |k| {
        bases[k] = sum;
        sum += counts[k];
    }
    pool.runChunks(&j, FilterJob.writeJob, nk);
    return sum;
}

/// Two passes: per-chunk counts, prefix sum, then every chunk writes into its own slice.
fn filter_mt(amt: &[f64], pool: &Pool, parts: usize) -> Vec<u32> {
    let n = amt.len();
    let ranges = chunk_ranges(n, parts);
    let counts: Vec<usize> = pool.map(parts, |p| {
        let (a, b) = ranges[p];
        amt[a..b].iter().map(|&x| (x > 500.0) as usize).sum()
    });
    let mut starts = vec![0usize; parts + 1];
    for p in 0..parts {
        starts[p + 1] = starts[p] + counts[p];
    }
    let total = starts[parts];
    let mut out: Vec<u32> = Vec::with_capacity(total);
    let op = SyncPtr(out.as_mut_ptr());
    pool.run(parts, |p| unsafe {
        let (a, b) = ranges[p];
        let k = filter_range(amt, a, b, op.get().add(starts[p]), counts[p]);
        debug_assert_eq!(k, counts[p]);
    });
    unsafe { out.set_len(total) };
    out
}

/// Fills `out` (capacity n + 1: the branchless loop writes one slot past the last hit) with the selected row indices; returns the count.
func filterGt500(_ amt: UnsafePointer<Double>, _ n: Int, _ out: UnsafeMutablePointer<UInt32>, pool: Pool?, parts: Int) -> Int {
    guard let pool = pool, parts > 1 else { return filterRange(amt, 0, n, out) }
    let np = max(1, min(parts, n / 4096))
    let chunk = (n + np - 1) / np
    let counts = alloc(np + 1, Int.self), ends = alloc(np, Int.self)
    pool.run(np) { p in (counts[p], ends[p]) = countRange(amt, p * chunk, min(n, (p + 1) * chunk)) }   // pass 1: count
    var base = 0
    for p in 0..<np { let c = counts[p]; counts[p] = base; base += c }
    counts[np] = base
    pool.run(np) { p in _ = filterRange(amt, p * chunk, ends[p], out + counts[p]) }   // pass 2: write, stop after the last hit
    free(counts); free(ends)
    return base
}

func filterGT(amt []float64, thr float64, p *pool) []uint32 {
	n := len(amt)
	if p == nil {
		out := make([]uint32, n+1)
		k := 0
		for i, v := range amt {
			out[k] = uint32(i)
			if v > thr {
				k++
			}
		}
		return out[:k]
	}
	C := p.chunks()
	bases := make([]int, C+1)
	p.run(C, n, func(c, lo, hi int) {
		cnt := 0
		for _, v := range amt[lo:hi] {
			if v > thr {
				cnt++
			}
		}
		bases[c+1] = cnt
	})
	for c := 0; c < C; c++ {
		bases[c+1] += bases[c]
	}
	total := bases[C]
	out := make([]uint32, total+C)
	p.run(C, n, func(c, lo, hi int) {
		k, end, dummy := bases[c], bases[c+1], total+c
		for i := lo; i < hi; i++ {
			pos := k
			if pos >= end {
				pos = dummy
			}
			out[pos] = uint32(i)
			if amt[i] > thr {
				k++
			}
		}
	})
	return out[:total]
}

timed("filter", lambda: df.lazy().filter(pred).select(pl.len()).collect().item())

timed("filter", lambda: int((amount > 500).sum()))

timed("filter", lambda: con.execute("SELECT count(*) FROM t WHERE amount > 500").fetchone())

timed("filter", lambda: df.filter(pred).count())

 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;

// selection.h — the result of a filter: one bit per row. 1M rows = 125 KB,
// written at memory speed; indices are produced only when something needs them.

struct Bitmap {
    std::vector<uint64_t> words;
    size_t bits = 0;
    bool test(size_t i) const { return (words[i >> 6] >> (i & 63)) & 1; }
    size_t count() const;
    std::vector<uint32_t> indices() const;  // ascending row indices of the set bits
};

// JS measurefilter semantics: NaN never matches. Sequential below ~1M values; the pool otherwise.
Bitmap compareDoubles(const double* v, size_t n, Cmp op, double threshold, Pool* pool);