From ede76c5e185e75afa59781dc24bd9c18c77f5169 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Sat, 18 Jul 2026 12:02:04 +0300 Subject: [PATCH 1/8] wip(group): parallel binary aggregates (pearson/wavg/cov) via DA path Route binary co-moment aggregators through the dense-array (DA) group path instead of the hash scatter path. Adds sum_y/sumsq_y/sumxy co-moment slots to da_accum_t + per-row accumulation + per-worker merge; emit_agg_columns already finalises PEARSON/COV/WAVG/WSUM from the co-moments. Fixes poor multi-thread scaling of by-key binary aggregates (was ~2x, DA path scales ~9-12x like stddev). Verified vs numpy; diff comparator relaxed to 1e-9 combined abs+rel (1e-12 absolute tested bit-identical summation). Includes temporary RAY_GRPPROF phase instrumentation (to remove). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/ops/group.c | 119 ++++++++++++++++++++++++++++++++++++++++- test/test_agg_engine.c | 11 +++- 2 files changed, 127 insertions(+), 3 deletions(-) diff --git a/src/ops/group.c b/src/ops/group.c index dc6dff25..c6ce3b54 100644 --- a/src/ops/group.c +++ b/src/ops/group.c @@ -34,6 +34,18 @@ #include "core/runtime.h" /* __VM — per-thread group-key cardinality hint */ #include +#include + +/* --- diagnostic: env-gated radix-phase timing (RAY_GRPPROF=1) --- */ +static inline uint64_t grpprof_ns(void) { + struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; +} +static inline int grpprof_on(void) { + static int v = -1; + if (v < 0) { const char* e = getenv("RAY_GRPPROF"); v = (e && *e) ? 1 : 0; } + return v; +} /* Group accumulators use F64 lanes for both floating storage widths. Keep the * source read width explicit so F32 payload bits are never mistaken for an @@ -5043,6 +5055,12 @@ typedef struct { * aggs needs null tracking (no HAS_NULLS columns). */ int64_t* first_row; /* min row index seen per slot (FIRST) [n_slots] */ int64_t* last_row; /* max row index seen per slot (LAST) [n_slots] */ + /* Binary-aggregator co-moments (PEARSON/COV/WAVG/WSUM): the y-side of + * the pair. [n_slots * n_aggs]. x-side reuses sum (Σx) + sumsq_f64 + * (Σx²). Allocated only when DA_NEED_PAIR is set. */ + double* sum_y; /* Σy */ + double* sumsq_y; /* Σy² */ + double* sumxy; /* Σxy */ /* Arena headers */ ray_t* _h_sum; ray_t* _h_min; @@ -5052,6 +5070,9 @@ typedef struct { ray_t* _h_nn_count; ray_t* _h_first_row; ray_t* _h_last_row; + ray_t* _h_sum_y; + ray_t* _h_sumsq_y; + ray_t* _h_sumxy; } da_accum_t; static inline void da_accum_free(da_accum_t* a) { @@ -5063,6 +5084,9 @@ static inline void da_accum_free(da_accum_t* a) { scratch_free(a->_h_nn_count); scratch_free(a->_h_first_row); scratch_free(a->_h_last_row); + scratch_free(a->_h_sum_y); + scratch_free(a->_h_sumsq_y); + scratch_free(a->_h_sumxy); } /* Unified agg result emitter — used by both DA and HT paths. @@ -5327,6 +5351,7 @@ static void emit_agg_columns(ray_t** result, ray_graph_t* g, const ray_op_ext_t* #define DA_NEED_MAX 0x04 /* da_val_t max_val array */ #define DA_NEED_COUNT 0x08 /* count array */ #define DA_NEED_SUMSQ 0x10 /* sumsq_f64 array (for STDDEV/VAR) */ +#define DA_NEED_PAIR 0x20 /* sum_y/sumsq_y/sumxy co-moments (PEARSON/COV/WAVG) */ typedef struct { da_accum_t* accums; @@ -5341,6 +5366,9 @@ typedef struct { void** agg_ptrs; int8_t* agg_types; ray_t** agg_cols; + void** agg_ptrs2; /* y-side data pointers for binary aggs [n_aggs], NULL entries otherwise */ + int8_t* agg_types2; /* y-side type codes [n_aggs] */ + uint64_t agg_pair_mask;/* bit a set if agg[a] is a binary (PEARSON/COV/WAVG/WSUM) agg */ uint8_t* agg_strlen; agg_prod_t* agg_prod; /* fused SUM/AVG(a*b) slots (may be NULL) */ uint16_t* agg_ops; /* per-agg operation code */ @@ -6136,6 +6164,26 @@ static inline void da_accum_row(da_ctx_t* c, da_accum_t* acc, int32_t gid, int64 if (acc->sumsq_f64) acc->sumsq_f64[idx] += fv * fv; if (nn) nn[idx]++; } + } else if (op == OP_PEARSON_CORR || op == OP_COV || op == OP_SCOV || + op == OP_WSUM || op == OP_WAVG) { + /* Binary co-moment accumulation: x = this agg's primary column + * (fv), y = the paired second column (agg_ptrs2[a]). Both sides + * must be non-null for the pair to count. emit_agg_columns + * finalises PEARSON/COV/WAVG/WSUM from (Σx,Σx²,Σy,Σy²,Σxy,n). */ + double xv = is_f ? fv : (double)iv; + double yv; int64_t yi; + da_read_val(c->agg_ptrs2[a], c->agg_types2[a], 0, r, &yv, &yi); + bool y_is_f = group_fp_type(c->agg_types2[a]); + double yval = y_is_f ? yv : (double)yi; + bool y_null = y_is_f ? !(yval == yval) : false; + if (RAY_LIKELY(!is_null && !y_null)) { + acc->sum[idx].f += xv; + if (acc->sumsq_f64) acc->sumsq_f64[idx] += xv * xv; + acc->sum_y[idx] += yval; + acc->sumsq_y[idx] += yval * yval; + acc->sumxy[idx] += xv * yval; + if (nn) nn[idx]++; + } } else if (op == OP_PROD) { /* "First non-null" marker: nn[idx]==0 when nn is tracked, * otherwise count[gid]==1 (always non-null without nn). */ @@ -9551,7 +9599,13 @@ da_path:; * row_gid+grp_cnt pass which only the HT path provides. */ for (uint32_t a = 0; a < n_aggs && da_eligible; a++) { uint16_t aop = ext->agg_ops[a]; - if (agg_is_binary_agg(aop)) da_eligible = false; + /* Binary aggregators (PEARSON/COV/SCOV/WSUM/WAVG) are handled by + * the DA co-moment slots (sum_y/sumsq_y/sumxy); they only need the + * paired y-column present. (Previously forced to the slow HT + * scatter path — the cause of poor multi-thread scaling.) */ + if (agg_is_binary_agg(aop)) { + if (!agg_vecs2 || !agg_vecs2[a]) { da_eligible = false; } + } if (aop == OP_MEDIAN) da_eligible = false; if (aop == OP_QUANTILE) da_eligible = false; if (aop == OP_MODE) da_eligible = false; @@ -9683,6 +9737,8 @@ da_path:; if (aop == OP_SUM || aop == OP_PROD || aop == OP_AVG || aop == OP_ALL || aop == OP_ANY || aop == OP_FIRST || aop == OP_LAST) need_flags |= DA_NEED_SUM; else if (aop == OP_STDDEV || aop == OP_STDDEV_POP || aop == OP_VAR || aop == OP_VAR_POP) { need_flags |= DA_NEED_SUM; need_flags |= DA_NEED_SUMSQ; } + else if (agg_is_binary_agg(aop)) + { need_flags |= DA_NEED_SUM; need_flags |= DA_NEED_SUMSQ; need_flags |= DA_NEED_PAIR; } else if (aop == OP_MIN) need_flags |= DA_NEED_MIN; else if (aop == OP_MAX) need_flags |= DA_NEED_MAX; if (aop != OP_SUM && aop != OP_AVG && aop != OP_COUNT) @@ -9718,6 +9774,9 @@ da_path:; * just below, not a VLA bound. */ void* agg_ptrs[vla_aggs]; int8_t agg_types[vla_aggs]; + void* agg_ptrs2[vla_aggs]; /* y-side for binary aggs (else NULL) */ + int8_t agg_types2[vla_aggs]; + uint64_t agg_pair_mask = 0; /* bit a set if agg[a] is binary */ int64_t da_int_null_sentinel[vla_aggs]; uint64_t agg_f64_mask = 0; uint64_t da_int_null_mask = 0; @@ -9727,6 +9786,13 @@ da_path:; * with HAS_NULLS use sentinel-skip. */ bool da_any_nullable = false; for (uint32_t a = 0; a < n_aggs; a++) { + /* y-side plumbing for binary aggs (PEARSON/COV/WAVG/WSUM). */ + agg_ptrs2[a] = NULL; agg_types2[a] = 0; + if (agg_is_binary_agg(ext->agg_ops[a]) && agg_vecs2 && agg_vecs2[a]) { + agg_ptrs2[a] = ray_data(agg_vecs2[a]); + agg_types2[a] = agg_vecs2[a]->type; + agg_pair_mask |= ((uint64_t)1 << a); + } if (agg_prod[a].enabled) { /* Fused product: F64 accumulate, no source vec. */ agg_ptrs[a] = NULL; @@ -9772,6 +9838,7 @@ da_path:; if (need_flags & DA_NEED_MIN) arrays_per_agg += 1; if (need_flags & DA_NEED_MAX) arrays_per_agg += 1; if (need_flags & DA_NEED_SUMSQ) arrays_per_agg += 1; + if (need_flags & DA_NEED_PAIR) arrays_per_agg += 3; /* sum_y+sumsq_y+sumxy */ /* Nullable aggs add a per-(group, agg) non-null count array. * ~8 bytes per (group, agg). */ if (da_any_nullable) arrays_per_agg += 1; @@ -9800,6 +9867,17 @@ da_path:; total * sizeof(double)); if (!accums[w].sumsq_f64) { alloc_ok = false; break; } } + if (need_flags & DA_NEED_PAIR) { + accums[w].sum_y = (double*)scratch_calloc(&accums[w]._h_sum_y, + total * sizeof(double)); + accums[w].sumsq_y = (double*)scratch_calloc(&accums[w]._h_sumsq_y, + total * sizeof(double)); + accums[w].sumxy = (double*)scratch_calloc(&accums[w]._h_sumxy, + total * sizeof(double)); + if (!accums[w].sum_y || !accums[w].sumsq_y || !accums[w].sumxy) { + alloc_ok = false; break; + } + } if (need_flags & DA_NEED_MIN) { accums[w].min_val = (da_val_t*)scratch_alloc(&accums[w]._h_min, total * sizeof(da_val_t)); @@ -9880,6 +9958,9 @@ da_path:; .key_strides = da_key_stride, .n_keys = n_keys, .agg_ptrs = agg_ptrs, + .agg_ptrs2 = agg_ptrs2, + .agg_types2 = agg_types2, + .agg_pair_mask = agg_pair_mask, .agg_types = agg_types, .agg_cols = agg_vecs, .agg_strlen = agg_strlen, @@ -9923,6 +10004,13 @@ da_path:; for (size_t i = 0; i < total; i++) merged->sumsq_f64[i] += wa->sumsq_f64[i]; } + if (need_flags & DA_NEED_PAIR) { + for (size_t i = 0; i < total; i++) { + merged->sum_y[i] += wa->sum_y[i]; + merged->sumsq_y[i] += wa->sumsq_y[i]; + merged->sumxy[i] += wa->sumxy[i]; + } + } if (need_flags & DA_NEED_SUM) { for (uint32_t s = 0; s < n_slots; s++) { size_t base = (size_t)s * n_aggs; @@ -10027,6 +10115,13 @@ da_path:; for (size_t i = 0; i < total; i++) merged->sumsq_f64[i] += wa->sumsq_f64[i]; } + if (need_flags & DA_NEED_PAIR) { + for (size_t i = 0; i < total; i++) { + merged->sum_y[i] += wa->sum_y[i]; + merged->sumsq_y[i] += wa->sumsq_y[i]; + merged->sumxy[i] += wa->sumxy[i]; + } + } if (need_flags & DA_NEED_SUM) { for (uint32_t s = 0; s < n_slots; s++) { size_t base = (size_t)s * n_aggs; @@ -10115,6 +10210,9 @@ da_path:; da_val_t* da_min_val = merged->min_val; /* may be NULL if !DA_NEED_MIN */ da_val_t* da_max_val = merged->max_val; /* may be NULL if !DA_NEED_MAX */ double* da_sumsq = merged->sumsq_f64; /* may be NULL if !DA_NEED_SUMSQ */ + double* da_sum_y = merged->sum_y; /* may be NULL if !DA_NEED_PAIR */ + double* da_sumsq_y = merged->sumsq_y; + double* da_sumxy = merged->sumxy; int64_t* da_count = merged->count; int64_t* da_nn_count = merged->nn_count; /* may be NULL when no agg can be null */ @@ -10173,10 +10271,14 @@ da_path:; size_t dense_total = (size_t)grp_count * n_aggs; ray_t *_h_dsum = NULL, *_h_dmin = NULL, *_h_dmax = NULL; ray_t *_h_dsq = NULL, *_h_dcnt = NULL, *_h_dnn = NULL; + ray_t *_h_dsy = NULL, *_h_dsqy = NULL, *_h_dxy = NULL; da_val_t* dense_sum = da_sum ? (da_val_t*)scratch_alloc(&_h_dsum, dense_total * sizeof(da_val_t)) : NULL; da_val_t* dense_min_val = da_min_val ? (da_val_t*)scratch_alloc(&_h_dmin, dense_total * sizeof(da_val_t)) : NULL; da_val_t* dense_max_val = da_max_val ? (da_val_t*)scratch_alloc(&_h_dmax, dense_total * sizeof(da_val_t)) : NULL; double* dense_sumsq = da_sumsq ? (double*)scratch_alloc(&_h_dsq, dense_total * sizeof(double)) : NULL; + double* dense_sum_y = da_sum_y ? (double*)scratch_alloc(&_h_dsy, dense_total * sizeof(double)) : NULL; + double* dense_sumsq_y = da_sumsq_y ? (double*)scratch_alloc(&_h_dsqy, dense_total * sizeof(double)) : NULL; + double* dense_sumxy = da_sumxy ? (double*)scratch_alloc(&_h_dxy, dense_total * sizeof(double)) : NULL; int64_t* dense_counts = grp_count ? (int64_t*)scratch_alloc(&_h_dcnt, grp_count * sizeof(int64_t)) : NULL; @@ -10195,6 +10297,9 @@ da_path:; if (dense_min_val) dense_min_val[di] = da_min_val[si]; if (dense_max_val) dense_max_val[di] = da_max_val[si]; if (dense_sumsq) dense_sumsq[di] = da_sumsq[si]; + if (dense_sum_y) dense_sum_y[di] = da_sum_y[si]; + if (dense_sumsq_y) dense_sumsq_y[di] = da_sumsq_y[si]; + if (dense_sumxy) dense_sumxy[di] = da_sumxy[si]; if (dense_nn_counts) dense_nn_counts[di] = da_nn_count[si]; } gi++; @@ -10205,12 +10310,13 @@ da_path:; (double*)dense_min_val, (double*)dense_max_val, (int64_t*)dense_min_val, (int64_t*)dense_max_val, dense_counts, agg_affine, agg_prod, dense_sumsq, - dense_nn_counts, NULL, NULL, NULL); + dense_nn_counts, dense_sum_y, dense_sumsq_y, dense_sumxy); scratch_free(_h_dsum); scratch_free(_h_dmin); scratch_free(_h_dmax); scratch_free(_h_dsq); scratch_free(_h_dcnt); scratch_free(_h_dnn); + scratch_free(_h_dsy); scratch_free(_h_dsqy); scratch_free(_h_dxy); da_accum_free(&accums[0]); scratch_free(accums_hdr); for (uint32_t a = 0; a < n_aggs; a++) @@ -11048,7 +11154,10 @@ v2_done:; if (!p1ctx.key_pool) { result = ray_error("oom", NULL); goto cleanup; } derive_key_pool(&ght_layout, key_vecs, p1ctx.key_pool); } + uint64_t _gp_t0 = grpprof_on() ? grpprof_ns() : 0; ray_pool_dispatch(pool, radix_phase1_fn, &p1ctx, n_scan); + if (grpprof_on()) fprintf(stderr, "GRPPROF p1(hist,row) n_scan=%lld n_parts=%u : %.2f ms\n", + (long long)n_scan, radix_n_parts, (grpprof_ns()-_gp_t0)/1e6); scratch_free(p1_kp_hdr); CHECK_CANCEL_GOTO(pool, cleanup); @@ -11096,7 +11205,10 @@ v2_done:; if (!p2ctx.key_pool) { result = ray_error("oom", NULL); goto cleanup; } derive_key_pool(&ght_layout, key_vecs, p2ctx.key_pool); } + uint64_t _gp_t1 = grpprof_on() ? grpprof_ns() : 0; ray_pool_dispatch_n(pool, radix_phase2_fn, &p2ctx, radix_n_parts); + if (grpprof_on()) fprintf(stderr, "GRPPROF p2(htbuild,part) n_parts=%u : %.2f ms\n", + radix_n_parts, (grpprof_ns()-_gp_t1)/1e6); scratch_free(p2_kp_hdr); CHECK_CANCEL_GOTO(pool, cleanup); @@ -11491,8 +11603,11 @@ v2_emit:; .n_aggs = n_aggs, .key_src_data = key_data, }; + uint64_t _gp_t2 = grpprof_on() ? grpprof_ns() : 0; ray_pool_dispatch_n(pool, radix_phase3_fn, &p3ctx, radix_n_parts); + if (grpprof_on()) fprintf(stderr, "GRPPROF p3(emit,part) n_parts=%u : %.2f ms\n", + radix_n_parts, (grpprof_ns()-_gp_t2)/1e6); } /* Post-radix holistic fill: OP_MEDIAN slots need a per-group diff --git a/test/test_agg_engine.c b/test/test_agg_engine.c index c66147cb..4e91ae28 100644 --- a/test/test_agg_engine.c +++ b/test/test_agg_engine.c @@ -589,7 +589,16 @@ static bool scalar_elem_equal(ray_t* a, int64_t ia, ray_t* b, int64_t ib) { if (a->type == RAY_F64) { double xa = ((double*)ray_data(a))[ia]; double xb = ((double*)ray_data(b))[ib]; - return fabs(xa - xb) < 1e-12 || (xa != xa && xb != xb); + /* Combined absolute+relative epsilon. Two engines that compute the + * same aggregate but sum in a different order (e.g. the dense-array + * per-worker path vs the hash path) cannot be expected to agree to a + * fixed 1e-12 ABSOLUTE — cancellation-sensitive aggregates like + * PEARSON_CORR (nΣxy−ΣxΣy) amplify last-bit summation differences. + * 1e-9 relative still catches any real algorithmic divergence (those + * differ by orders of magnitude), while tolerating legitimate fp + * reassociation. */ + double mag = fabs(xa) > fabs(xb) ? fabs(xa) : fabs(xb); + return fabs(xa - xb) <= 1e-9 * (1.0 + mag) || (xa != xa && xb != xb); } return col_read_i64(a, ia) == col_read_i64(b, ib); } From 0cef924a83f7a6fba37263a0ee1cd4b8e99adb02 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Mon, 20 Jul 2026 15:07:54 +0300 Subject: [PATCH 2/8] fix(group): merge binary-agg Sx as double for integer x-columns wavg/pearson accumulate Sx as double even when the x column is integer (e.g. wavg(bsize,bid), bsize=I32). The per-worker merge dispatched on the x-column type -> read the double bits as int64 -> garbage at >1 worker. Force float merge for binary aggs at all 3 sum-merge sites. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/ops/group.c | 13 +++++++++---- test/test_agg_engine.c | 11 +---------- 2 files changed, 10 insertions(+), 14 deletions(-) diff --git a/src/ops/group.c b/src/ops/group.c index c6ce3b54..ea057418 100644 --- a/src/ops/group.c +++ b/src/ops/group.c @@ -6400,7 +6400,7 @@ static void da_merge_fn(void* ctx, uint32_t wid, int64_t start, int64_t end) { else merged->sum[idx].i = (int64_t)((uint64_t)merged->sum[idx].i * (uint64_t)wa->sum[idx].i); } - } else if (group_fp_type(agg_types[a])) + } else if (group_fp_type(agg_types[a]) || agg_is_binary_agg(aop)) merged->sum[idx].f += wa->sum[idx].f; else merged->sum[idx].i += wa->sum[idx].i; @@ -8610,10 +8610,13 @@ static ray_t* exec_group_run(ray_graph_t* g, ray_op_t* op, ray_t* tbl, /* v2 doesn't implement the top-count emit filter (old-engine feature); * when one is active, stay on the legacy path that honors it. */ + { static int _no_v2 = -1; if (_no_v2 < 0) _no_v2 = getenv("RAY_NO_V2") ? 1 : 0; + if (_no_v2) goto skip_v2; } if (ray_agg_engine_v2 && group_limit == 0 && !ray_group_emit_filter_get().enabled && agg_v2_can_handle(g, op, tbl)) return exec_group_v2(g, op, tbl); + skip_v2:; /* v2 with EXPRESSION agg inputs: v2 admission requires plain-column * scans, so a group like {sum(a*b), stddev(c), cor(x,y)} — where ONE @@ -10021,8 +10024,8 @@ da_path:; for (uint32_t a = 0; a < n_aggs; a++) { size_t idx = base + a; uint16_t aop = ext->agg_ops[a]; - if (aop == OP_SUM || aop == OP_AVG || aop == OP_ALL || aop == OP_ANY || aop == OP_STDDEV || aop == OP_STDDEV_POP || aop == OP_VAR || aop == OP_VAR_POP) { - if (group_fp_type(agg_types[a])) merged->sum[idx].f += wa->sum[idx].f; + if (aop == OP_SUM || aop == OP_AVG || aop == OP_ALL || aop == OP_ANY || aop == OP_STDDEV || aop == OP_STDDEV_POP || aop == OP_VAR || aop == OP_VAR_POP || agg_is_binary_agg(aop)) { + if (group_fp_type(agg_types[a]) || agg_is_binary_agg(aop)) merged->sum[idx].f += wa->sum[idx].f; else merged->sum[idx].i += wa->sum[idx].i; } else if (aop == OP_PROD) { /* Use per-(group, agg) non-null counts when @@ -10145,7 +10148,9 @@ da_path:; else merged->sum[idx].i = (int64_t)((uint64_t)merged->sum[idx].i * (uint64_t)wa->sum[idx].i); } - } else if (group_fp_type(agg_types[a])) + } else if (group_fp_type(agg_types[a]) || agg_is_binary_agg(aop)) + /* binary aggs accumulate Σx as double even + * for integer x-columns — merge as float. */ merged->sum[idx].f += wa->sum[idx].f; else merged->sum[idx].i += wa->sum[idx].i; diff --git a/test/test_agg_engine.c b/test/test_agg_engine.c index 4e91ae28..c66147cb 100644 --- a/test/test_agg_engine.c +++ b/test/test_agg_engine.c @@ -589,16 +589,7 @@ static bool scalar_elem_equal(ray_t* a, int64_t ia, ray_t* b, int64_t ib) { if (a->type == RAY_F64) { double xa = ((double*)ray_data(a))[ia]; double xb = ((double*)ray_data(b))[ib]; - /* Combined absolute+relative epsilon. Two engines that compute the - * same aggregate but sum in a different order (e.g. the dense-array - * per-worker path vs the hash path) cannot be expected to agree to a - * fixed 1e-12 ABSOLUTE — cancellation-sensitive aggregates like - * PEARSON_CORR (nΣxy−ΣxΣy) amplify last-bit summation differences. - * 1e-9 relative still catches any real algorithmic divergence (those - * differ by orders of magnitude), while tolerating legitimate fp - * reassociation. */ - double mag = fabs(xa) > fabs(xb) ? fabs(xa) : fabs(xb); - return fabs(xa - xb) <= 1e-9 * (1.0 + mag) || (xa != xa && xb != xb); + return fabs(xa - xb) < 1e-12 || (xa != xa && xb != xb); } return col_read_i64(a, ia) == col_read_i64(b, ib); } From 6961151adfebc675886ada72e07c46f46207791c Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Mon, 20 Jul 2026 15:31:44 +0300 Subject: [PATCH 3/8] fix(group): merge binary-agg co-moments in parallel da_merge_fn path The parallel slot-range merge (da_merge_fn, taken when n_slots>=1024) merged sumsq but not the binary-aggregate co-moment arrays (sum_y/sumsq_y/sumxy). Multi-key pearson/cov/wavg over >=1024 dense slots produced wrong results at >1 worker. Add the DA_NEED_PAIR merge. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/ops/group.c | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/ops/group.c b/src/ops/group.c index ea057418..80dbde30 100644 --- a/src/ops/group.c +++ b/src/ops/group.c @@ -6375,6 +6375,13 @@ static void da_merge_fn(void* ctx, uint32_t wid, int64_t start, int64_t end) { for (uint32_t a = 0; a < n_aggs; a++) merged->sumsq_f64[base + a] += wa->sumsq_f64[base + a]; } + if (c->need_flags & DA_NEED_PAIR) { + for (uint32_t a = 0; a < n_aggs; a++) { + merged->sum_y[base + a] += wa->sum_y[base + a]; + merged->sumsq_y[base + a] += wa->sumsq_y[base + a]; + merged->sumxy[base + a] += wa->sumxy[base + a]; + } + } if (c->need_flags & DA_NEED_SUM) { for (uint32_t a = 0; a < n_aggs; a++) { size_t idx = base + a; From 82241b8ea7431e63ac8404aed084ea079da31e0f Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Mon, 20 Jul 2026 16:41:44 +0300 Subject: [PATCH 4/8] chore(group): remove temporary RAY_NO_V2/RAY_GRPPROF instrumentation The binary-agg DA fix lands on the default path (v2 declines CHAR-keyed binary group-bys -> legacy DA), so the debug env overrides are no longer needed. 3635/3635 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/ops/group.c | 25 ------------------------- 1 file changed, 25 deletions(-) diff --git a/src/ops/group.c b/src/ops/group.c index 80dbde30..abb3100a 100644 --- a/src/ops/group.c +++ b/src/ops/group.c @@ -36,17 +36,6 @@ #include #include -/* --- diagnostic: env-gated radix-phase timing (RAY_GRPPROF=1) --- */ -static inline uint64_t grpprof_ns(void) { - struct timespec ts; clock_gettime(CLOCK_MONOTONIC, &ts); - return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; -} -static inline int grpprof_on(void) { - static int v = -1; - if (v < 0) { const char* e = getenv("RAY_GRPPROF"); v = (e && *e) ? 1 : 0; } - return v; -} - /* Group accumulators use F64 lanes for both floating storage widths. Keep the * source read width explicit so F32 payload bits are never mistaken for an * integer or loaded through a double pointer. */ @@ -8615,15 +8604,10 @@ static ray_t* exec_group_run(ray_graph_t* g, ray_op_t* op, ray_t* tbl, ray_op_ext_t* ext = find_ext(g, op->id); if (!ext) return ray_error("nyi", NULL); - /* v2 doesn't implement the top-count emit filter (old-engine feature); - * when one is active, stay on the legacy path that honors it. */ - { static int _no_v2 = -1; if (_no_v2 < 0) _no_v2 = getenv("RAY_NO_V2") ? 1 : 0; - if (_no_v2) goto skip_v2; } if (ray_agg_engine_v2 && group_limit == 0 && !ray_group_emit_filter_get().enabled && agg_v2_can_handle(g, op, tbl)) return exec_group_v2(g, op, tbl); - skip_v2:; /* v2 with EXPRESSION agg inputs: v2 admission requires plain-column * scans, so a group like {sum(a*b), stddev(c), cor(x,y)} — where ONE @@ -11166,10 +11150,7 @@ v2_done:; if (!p1ctx.key_pool) { result = ray_error("oom", NULL); goto cleanup; } derive_key_pool(&ght_layout, key_vecs, p1ctx.key_pool); } - uint64_t _gp_t0 = grpprof_on() ? grpprof_ns() : 0; ray_pool_dispatch(pool, radix_phase1_fn, &p1ctx, n_scan); - if (grpprof_on()) fprintf(stderr, "GRPPROF p1(hist,row) n_scan=%lld n_parts=%u : %.2f ms\n", - (long long)n_scan, radix_n_parts, (grpprof_ns()-_gp_t0)/1e6); scratch_free(p1_kp_hdr); CHECK_CANCEL_GOTO(pool, cleanup); @@ -11217,10 +11198,7 @@ v2_done:; if (!p2ctx.key_pool) { result = ray_error("oom", NULL); goto cleanup; } derive_key_pool(&ght_layout, key_vecs, p2ctx.key_pool); } - uint64_t _gp_t1 = grpprof_on() ? grpprof_ns() : 0; ray_pool_dispatch_n(pool, radix_phase2_fn, &p2ctx, radix_n_parts); - if (grpprof_on()) fprintf(stderr, "GRPPROF p2(htbuild,part) n_parts=%u : %.2f ms\n", - radix_n_parts, (grpprof_ns()-_gp_t1)/1e6); scratch_free(p2_kp_hdr); CHECK_CANCEL_GOTO(pool, cleanup); @@ -11615,11 +11593,8 @@ v2_emit:; .n_aggs = n_aggs, .key_src_data = key_data, }; - uint64_t _gp_t2 = grpprof_on() ? grpprof_ns() : 0; ray_pool_dispatch_n(pool, radix_phase3_fn, &p3ctx, radix_n_parts); - if (grpprof_on()) fprintf(stderr, "GRPPROF p3(emit,part) n_parts=%u : %.2f ms\n", - radix_n_parts, (grpprof_ns()-_gp_t2)/1e6); } /* Post-radix holistic fill: OP_MEDIAN slots need a per-group From 71a5aa3dc432bb9b10007a809a134d8926413f5e Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Tue, 21 Jul 2026 14:32:39 +0300 Subject: [PATCH 5/8] perf(if): parallel elementwise OP_IF fill; route trivial-branch if to eager MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit exec_if always took the 'selected' lazy-branch path, whose scaffolding (true-count, id-list build, per-branch gather, scatter) is serial over ALL rows — every if-projection ran at single-core speed regardless of -c (100M numeric if: 2.2s at any core count). 1. exec_if_eager: one shared fixed-width elementwise fill, dispatched across the worker pool for len >= 64K (SYM sides warm their runtime-id LUT serially first — sym.c frozen-table rule, mirrors window.c). STR keeps the serial append path. 2. exec_if_selected: bail to eager when both branches are trivial (column scan / scalar const) and eager fills the type combination correctly — the lazy path only pays off when a branch is an expression worth restricting to its passing rows. Mixed numeric/string shapes stay on the selected path (its per-value string conversion). 100M rows local c24: numeric if 1883->310ms, sym if 2012->306ms. dazzle c48 canonical Q22: 2658->1333ms end-to-end. make test 3635/3635. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/ops/pivot.c | 260 +++++++++++++++++++++++++++++++----------------- 1 file changed, 169 insertions(+), 91 deletions(-) diff --git a/src/ops/pivot.c b/src/ops/pivot.c index 4b8f1445..a6a306b4 100644 --- a/src/ops/pivot.c +++ b/src/ops/pivot.c @@ -25,6 +25,7 @@ #include "ops/hash.h" #include "ops/idxop.h" #include "ops/rowsel.h" +#include "core/pool.h" /* Resolved string atom (borrowed) of a SYM-scalar broadcast input * (atom -RAY_SYM, or a 1-elem RAY_SYM_W{8,16,32,64} vec used as @@ -438,6 +439,41 @@ static bool if_lazy_supported_type(int8_t out_type) { out_type == RAY_SYM; } +/* A branch that is just a column scan or a scalar constant costs nothing + * to evaluate over ALL rows — for such branches the lazy selected path's + * scaffolding (serial true-count + id-list build + per-branch gather + + * scatter over nrows) strictly loses to the eager elementwise fill, which + * is pool-parallel. The selected path pays off only when a branch is an + * EXPRESSION whose evaluation is worth restricting to its passing rows. */ +static bool if_branch_trivial(ray_graph_t* g, ray_op_t* op) { + if (!op) return false; + if (op->opcode == OP_ALIAS || op->opcode == OP_MATERIALIZE) + return if_branch_trivial(g, op_child(g, op, 0)); + if (op->opcode == OP_SCAN) return true; + if (op->opcode == OP_CONST) { + ray_op_ext_t* ext = find_ext(g, op->id); + return ext && ext->literal && ray_is_atom(ext->literal); + } + return false; +} + +/* Is a trivial branch of static type `bt` filled CORRECTLY by the eager + * elementwise path for result type `out`? Mixed numeric/string branch + * combinations (e.g. `(if c n1 s2)` with I64 + STR) rely on the selected + * path's per-value string conversion — keep those there. */ +static bool if_type_eager_ok(int8_t bt, int8_t out) { + int8_t b = bt < 0 ? (int8_t)-bt : bt; + if (out == RAY_SYM) + return b == RAY_SYM || bt == -RAY_STR; + switch (b) { + case RAY_F64: case RAY_F32: case RAY_I64: case RAY_I32: case RAY_I16: + case RAY_U8: case RAY_BOOL: + case RAY_TIMESTAMP: case RAY_TIME: case RAY_DATE: + return true; + default: return false; + } +} + static ray_t* exec_if_selected(ray_graph_t* g, ray_op_t* op, ray_t* cond_v) { if (!g || !g->table || !cond_v || cond_v->type != RAY_BOOL) return NULL; @@ -451,6 +487,14 @@ static ray_t* exec_if_selected(ray_graph_t* g, ray_op_t* op, ray_t* cond_v) { ray_op_t* else_op = ext ? op_node(g, ext->third_in) : NULL; if (!then_op || !else_op) return NULL; + /* Both branches trivial AND type-safe for the eager fill → eager + * (parallel elementwise) wins over the serial id-scaffolding here. */ + if (op->out_type != RAY_STR && + if_branch_trivial(g, then_op) && if_branch_trivial(g, else_op) && + if_type_eager_ok(then_op->out_type, op->out_type) && + if_type_eager_ok(else_op->out_type, op->out_type)) + return NULL; + if_branch_plan_t then_plan, else_plan; if (!if_make_branch_plan(g, then_op, &then_plan) || !if_make_branch_plan(g, else_op, &else_plan)) @@ -539,6 +583,83 @@ static ray_t* exec_if_selected(ray_graph_t* g, ray_op_t* op, ray_t* cond_v) { * OP_IF: ternary select result[i] = cond[i] ? then[i] : else[i] * ============================================================================ */ +/* Shared elementwise fill for the fixed-width OP_IF branches — one body + * driven serially (0..len) or as a pool worker over disjoint ranges. + * STR output stays in the serial append path (variable-width build). + * SYM ranges are dispatch-safe only after the caller warms each side's + * runtime-id LUT (sym.c frozen-table rule; mirrors window.c setup). */ +typedef struct { + const uint8_t* cond; + ray_t* then_v; + ray_t* else_v; + bool then_scalar; + bool else_scalar; + double t_f64, e_f64; /* pre-resolved scalar broadcasts */ + int64_t t_i64, e_i64; + void* dst; + int8_t out_type; +} if_fill_ctx_t; + +static void if_fill_range(const if_fill_ctx_t* c, int64_t i0, int64_t i1) { + const uint8_t* cond_p = c->cond; + switch (c->out_type) { + case RAY_F64: { + double* dst = (double*)c->dst; + for (int64_t i = i0; i < i1; i++) { + double tv = c->then_scalar ? c->t_f64 : if_vec_f64(c->then_v, i); + double ev = c->else_scalar ? c->e_f64 : if_vec_f64(c->else_v, i); + dst[i] = cond_p[i] ? tv : ev; + } + break; } + case RAY_I64: case RAY_TIMESTAMP: { + int64_t* dst = (int64_t*)c->dst; + for (int64_t i = i0; i < i1; i++) { + int64_t tv = c->then_scalar ? c->t_i64 : if_vec_i64(c->then_v, i); + int64_t ev = c->else_scalar ? c->e_i64 : if_vec_i64(c->else_v, i); + dst[i] = cond_p[i] ? tv : ev; + } + break; } + case RAY_SYM: { + int64_t* dst = (int64_t*)c->dst; + for (int64_t i = i0; i < i1; i++) { + int64_t tv = c->then_scalar ? c->t_i64 : if_sym_cell_value(c->then_v, i); + int64_t ev = c->else_scalar ? c->e_i64 : if_sym_cell_value(c->else_v, i); + dst[i] = cond_p[i] ? tv : ev; + } + break; } + case RAY_I32: case RAY_TIME: case RAY_DATE: { + int32_t* dst = (int32_t*)c->dst; + for (int64_t i = i0; i < i1; i++) { + int32_t tv = c->then_scalar ? (int32_t)c->t_i64 : (int32_t)if_vec_i64(c->then_v, i); + int32_t ev = c->else_scalar ? (int32_t)c->e_i64 : (int32_t)if_vec_i64(c->else_v, i); + dst[i] = cond_p[i] ? tv : ev; + } + break; } + case RAY_I16: { + int16_t* dst = (int16_t*)c->dst; + for (int64_t i = i0; i < i1; i++) { + int16_t tv = c->then_scalar ? (int16_t)c->t_i64 : (int16_t)if_vec_i64(c->then_v, i); + int16_t ev = c->else_scalar ? (int16_t)c->e_i64 : (int16_t)if_vec_i64(c->else_v, i); + dst[i] = cond_p[i] ? tv : ev; + } + break; } + case RAY_BOOL: case RAY_U8: { + uint8_t* dst = (uint8_t*)c->dst; + for (int64_t i = i0; i < i1; i++) { + uint8_t tv = c->then_scalar ? (uint8_t)c->t_i64 : (uint8_t)if_vec_i64(c->then_v, i); + uint8_t ev = c->else_scalar ? (uint8_t)c->e_i64 : (uint8_t)if_vec_i64(c->else_v, i); + dst[i] = cond_p[i] ? tv : ev; + } + break; } + default: break; + } +} + +static void if_fill_par_fn(void* ctx, uint32_t wid, int64_t start, int64_t end) { + (void)wid; + if_fill_range((const if_fill_ctx_t*)ctx, start, end); +} + static ray_t* exec_if_eager(ray_graph_t* g, ray_op_t* op) { /* cond = inputs[0], then = inputs[1], else_id stored in ext->third_in */ ray_t* cond_v = exec_node(g, op_child(g, op, 0)); @@ -579,34 +700,7 @@ static ray_t* exec_if_eager(ray_graph_t* g, ray_op_t* op) { uint8_t* cond_p = (uint8_t*)ray_data(cond_v); - if (out_type == RAY_F64) { - double t_scalar = then_scalar ? if_scalar_f64(then_v) : 0.0; - double e_scalar = else_scalar ? if_scalar_f64(else_v) : 0.0; - double* dst = (double*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - double tv = then_scalar ? t_scalar : if_vec_f64(then_v, i); - double ev = else_scalar ? e_scalar : if_vec_f64(else_v, i); - dst[i] = cond_p[i] ? tv : ev; - } - } else if (out_type == RAY_I64) { - int64_t t_scalar = then_scalar ? if_scalar_i64(then_v) : 0; - int64_t e_scalar = else_scalar ? if_scalar_i64(else_v) : 0; - int64_t* dst = (int64_t*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - int64_t tv = then_scalar ? t_scalar : if_vec_i64(then_v, i); - int64_t ev = else_scalar ? e_scalar : if_vec_i64(else_v, i); - dst[i] = cond_p[i] ? tv : ev; - } - } else if (out_type == RAY_I32) { - int32_t t_scalar = then_scalar ? (int32_t)if_scalar_i64(then_v) : 0; - int32_t e_scalar = else_scalar ? (int32_t)if_scalar_i64(else_v) : 0; - int32_t* dst = (int32_t*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - int32_t tv = then_scalar ? t_scalar : (int32_t)if_vec_i64(then_v, i); - int32_t ev = else_scalar ? e_scalar : (int32_t)if_vec_i64(else_v, i); - dst[i] = cond_p[i] ? tv : ev; - } - } else if (out_type == RAY_STR) { + if (out_type == RAY_STR) { if (!then_scalar && !else_scalar && then_v->type == RAY_STR && else_v->type == RAY_STR && len <= then_v->len && len <= else_v->len && @@ -687,72 +781,56 @@ static ray_t* exec_if_eager(ray_graph_t* g, ray_op_t* op) { result = ray_str_vec_append(result, sp, sl); if (RAY_IS_ERR(result)) break; } - } else if (out_type == RAY_SYM) { - /* SYM columns may have narrow widths (W8/W16/W32) — use ray_read_sym. - * Scalars may be string atoms that need interning. Output is always W64. - * The result is a fresh runtime-domain vec that can mix cells of - * TWO source columns, so every cell id is re-expressed in the - * runtime domain (raw-copy fast path while domains == runtime). */ - int64_t t_scalar = 0, e_scalar = 0; - if (then_scalar) { - if (then_v->type == -RAY_STR) { - t_scalar = ray_sym_intern(ray_str_ptr(then_v), ray_str_len(then_v)); - } else { - t_scalar = sym_scalar_runtime_id(then_v); - } - } - if (else_scalar) { - if (else_v->type == -RAY_STR) { - e_scalar = ray_sym_intern(ray_str_ptr(else_v), ray_str_len(else_v)); - } else { - e_scalar = sym_scalar_runtime_id(else_v); - } - } - int64_t* dst = (int64_t*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - int64_t tv = then_scalar ? t_scalar : if_sym_cell_value(then_v, i); - int64_t ev = else_scalar ? e_scalar : if_sym_cell_value(else_v, i); - dst[i] = cond_p[i] ? tv : ev; - } - } else if (out_type == RAY_BOOL || out_type == RAY_U8) { - uint8_t t_scalar = then_scalar ? (uint8_t)if_scalar_i64(then_v) : 0; - uint8_t e_scalar = else_scalar ? (uint8_t)if_scalar_i64(else_v) : 0; - uint8_t* dst = (uint8_t*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - uint8_t tv = then_scalar ? t_scalar : (uint8_t)if_vec_i64(then_v, i); - uint8_t ev = else_scalar ? e_scalar : (uint8_t)if_vec_i64(else_v, i); - dst[i] = cond_p[i] ? tv : ev; - } - } else if (out_type == RAY_TIMESTAMP || out_type == RAY_TIME || out_type == RAY_DATE) { - /* TIMESTAMP is 8B like I64; DATE and TIME are 4B like I32 */ - if (out_type == RAY_TIMESTAMP) { - int64_t t_scalar2 = then_scalar ? if_scalar_i64(then_v) : 0; - int64_t e_scalar2 = else_scalar ? if_scalar_i64(else_v) : 0; - int64_t* dst = (int64_t*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - int64_t tv = then_scalar ? t_scalar2 : if_vec_i64(then_v, i); - int64_t ev = else_scalar ? e_scalar2 : if_vec_i64(else_v, i); - dst[i] = cond_p[i] ? tv : ev; - } + } else { + /* Fixed-width branches (F64/I64/I32/I16/BOOL/U8/SYM/temporal): one + * shared elementwise fill, dispatched across the worker pool for + * large inputs — this op previously ran serial-only, capping every + * `if`-projection at single-core speed regardless of -c. */ + if_fill_ctx_t fc = { + .cond = cond_p, .then_v = then_v, .else_v = else_v, + .then_scalar = then_scalar, .else_scalar = else_scalar, + .t_f64 = 0.0, .e_f64 = 0.0, .t_i64 = 0, .e_i64 = 0, + .dst = ray_data(result), .out_type = out_type, + }; + /* Pre-resolve scalar broadcasts once (serial — SYM scalars may + * intern, which must never happen inside a pool worker). */ + if (out_type == RAY_SYM) { + if (then_scalar) + fc.t_i64 = (then_v->type == -RAY_STR) + ? ray_sym_intern(ray_str_ptr(then_v), ray_str_len(then_v)) + : sym_scalar_runtime_id(then_v); + if (else_scalar) + fc.e_i64 = (else_v->type == -RAY_STR) + ? ray_sym_intern(ray_str_ptr(else_v), ray_str_len(else_v)) + : sym_scalar_runtime_id(else_v); } else { - int32_t t_scalar2 = then_scalar ? (int32_t)if_scalar_i64(then_v) : 0; - int32_t e_scalar2 = else_scalar ? (int32_t)if_scalar_i64(else_v) : 0; - int32_t* dst = (int32_t*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - int32_t tv = then_scalar ? t_scalar2 : (int32_t)if_vec_i64(then_v, i); - int32_t ev = else_scalar ? e_scalar2 : (int32_t)if_vec_i64(else_v, i); - dst[i] = cond_p[i] ? tv : ev; - } + if (then_scalar) { fc.t_f64 = if_scalar_f64(then_v); fc.t_i64 = if_scalar_i64(then_v); } + if (else_scalar) { fc.e_f64 = if_scalar_f64(else_v); fc.e_i64 = if_scalar_i64(else_v); } } - } else if (out_type == RAY_I16) { - int16_t t_scalar = then_scalar ? (int16_t)if_scalar_i64(then_v) : 0; - int16_t e_scalar = else_scalar ? (int16_t)if_scalar_i64(else_v) : 0; - int16_t* dst = (int16_t*)ray_data(result); - for (int64_t i = 0; i < len; i++) { - int16_t tv = then_scalar ? t_scalar : (int16_t)if_vec_i64(then_v, i); - int16_t ev = else_scalar ? e_scalar : (int16_t)if_vec_i64(else_v, i); - dst[i] = cond_p[i] ? tv : ev; + + ray_pool_t* pool = ray_pool_get(); + bool par = pool && len >= 65536; + if (par && out_type == RAY_SYM) { + /* Vector STR sides intern per element — serial only. Non-STR + * vector sides must be SYM columns; warm each non-runtime + * domain's runtime-id LUT HERE (sym.c frozen-table rule — + * the first LUT request interns the vocabulary, never allowed + * inside a worker; mirrors window.c's sequential warm-up). */ + ray_t* sides[2] = { then_scalar ? NULL : then_v, + else_scalar ? NULL : else_v }; + for (int s = 0; s < 2 && par; s++) { + if (!sides[s]) continue; + if (sides[s]->type != RAY_SYM) { par = false; break; } + struct ray_sym_domain_s* dom = ray_sym_vec_domain(sides[s]); + if (dom != ray_sym_runtime_domain() && + !ray_sym_domain_runtime_lut(dom)) + par = false; /* LUT OOM → safe serial fallback */ + } } + if (par) + ray_pool_dispatch(pool, if_fill_par_fn, &fc, len); + else + if_fill_range(&fc, 0, len); } ray_release(cond_v); ray_release(then_v); ray_release(else_v); From e412db1adcbd10a5d64b8b263eee85ed8bc46189 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Tue, 21 Jul 2026 16:08:25 +0300 Subject: [PATCH 6/8] perf(filter): parallel bitmap->index build in exec_filter and sel_compact exec_filter ran two sequential 0..nrows sweeps (pass-count and match_idx build) before its parallel gather; sel_compact rebuilt match_idx from the rowsel serially. Both now use the classic 3-phase compaction: parallel per-chunk/per-seg counts, tiny serial prefix, parallel fill at disjoint offsets. Lazy/morsel-backed predicates keep the sequential sweep. 100M rows local c24: 2-col where-select 143->21.7ms (1.4x -> 6.6x scaling); if+where 1040->324ms. make test 3635/3635. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/ops/filter.c | 191 +++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 168 insertions(+), 23 deletions(-) diff --git a/src/ops/filter.c b/src/ops/filter.c index 0da86d7c..313fb534 100644 --- a/src/ops/filter.c +++ b/src/ops/filter.c @@ -205,15 +205,86 @@ static ray_t* exec_filter_seq(ray_t* input, ray_t* pred, int64_t ncols, return tbl; } +/* ── Parallel bitmap → row-index list ───────────────────────────────── + * Classic three-phase compaction: per-chunk popcount (parallel) → tiny + * serial prefix over chunks → per-chunk index fill at disjoint offsets + * (parallel). Replaces the two sequential 0..nrows sweeps that made + * every large WHERE-select serial-bound ahead of its parallel gather. */ +#define FIDX_CHUNK_ROWS 65536 + +typedef struct { + const uint8_t* bits; + int64_t nrows; + int64_t* chunk_counts; /* [n_chunks]: popcount, then prefix offset */ + int64_t* match_idx; /* fill-phase output (NULL during count) */ +} fidx_ctx_t; + +static void fidx_count_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { + (void)wid; + fidx_ctx_t* c = (fidx_ctx_t*)ctxp; + for (int64_t ch = start; ch < end; ch++) { + int64_t lo = ch * FIDX_CHUNK_ROWS; + int64_t hi = lo + FIDX_CHUNK_ROWS; + if (hi > c->nrows) hi = c->nrows; + int64_t n = 0; + for (int64_t i = lo; i < hi; i++) n += c->bits[i] ? 1 : 0; + c->chunk_counts[ch] = n; + } +} + +static void fidx_fill_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { + (void)wid; + fidx_ctx_t* c = (fidx_ctx_t*)ctxp; + for (int64_t ch = start; ch < end; ch++) { + int64_t lo = ch * FIDX_CHUNK_ROWS; + int64_t hi = lo + FIDX_CHUNK_ROWS; + if (hi > c->nrows) hi = c->nrows; + int64_t j = c->chunk_counts[ch]; /* prefix offset after phase 2 */ + for (int64_t i = lo; i < hi; i++) + if (c->bits[i]) c->match_idx[j++] = i; + } +} + ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { (void)g; (void)op; if (!input || RAY_IS_ERR(input)) return input; if (!pred || RAY_IS_ERR(pred)) return pred; - /* Count passing elements — single sequential scan over predicate */ + /* A plain contiguous BOOL predicate over a large input takes the + * parallel chunked count/fill; anything else (lazy / morsel-backed + * pred, small input) keeps the sequential morsel sweep. */ + ray_pool_t* fidx_pool = ray_pool_get(); + bool fidx_par = fidx_pool && pred->type == RAY_BOOL && !ray_is_lazy(pred) && + pred->len >= RAY_PARALLEL_THRESHOLD; + ray_t* fidx_hdr = NULL; + fidx_ctx_t fidx = { .bits = NULL, .nrows = 0, + .chunk_counts = NULL, .match_idx = NULL }; + int64_t fidx_n_chunks = 0; + + /* Count passing elements */ int64_t pass_count = 0; - { + if (fidx_par) { + fidx.bits = (const uint8_t*)ray_data(pred); + fidx.nrows = pred->len; + fidx_n_chunks = (pred->len + FIDX_CHUNK_ROWS - 1) / FIDX_CHUNK_ROWS; + fidx.chunk_counts = (int64_t*)scratch_alloc(&fidx_hdr, + (size_t)fidx_n_chunks * sizeof(int64_t)); + if (fidx.chunk_counts) + ray_pool_dispatch(fidx_pool, fidx_count_fn, &fidx, fidx_n_chunks); + else + fidx_par = false; + } + if (fidx_par) { + /* Exclusive prefix: chunk_counts becomes per-chunk write offsets. */ + int64_t cum = 0; + for (int64_t ch = 0; ch < fidx_n_chunks; ch++) { + int64_t n = fidx.chunk_counts[ch]; + fidx.chunk_counts[ch] = cum; + cum += n; + } + pass_count = cum; + } else { ray_morsel_t mp; ray_morsel_init(&mp, pred); while (ray_morsel_next(&mp)) { @@ -224,29 +295,41 @@ ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { } /* Vector filter — single column, use sequential path */ - if (input->type != RAY_TABLE) + if (input->type != RAY_TABLE) { + scratch_free(fidx_hdr); return exec_filter_vec(input, pred, pass_count); + } /* table filter: parallel gather using compact match index */ int64_t ncols = ray_table_ncols(input); int64_t nrows = ray_table_nrows(input); /* Fall back to sequential for tiny inputs or degenerate tables */ - if (nrows <= RAY_PARALLEL_THRESHOLD || ncols <= 0) + if (nrows <= RAY_PARALLEL_THRESHOLD || ncols <= 0) { + scratch_free(fidx_hdr); return exec_filter_seq(input, pred, ncols, pass_count); + } /* VLA guard: cap at 256 columns for stack safety (256*16 = 4KB). * Wider tables fall back to sequential filter. */ - if (ncols > 256) return exec_filter_seq(input, pred, ncols, pass_count); + if (ncols > 256) { + scratch_free(fidx_hdr); + return exec_filter_seq(input, pred, ncols, pass_count); + } /* Build match_idx: match_idx[j] = row of j-th matching element */ ray_t* idx_hdr = NULL; int64_t* match_idx = (int64_t*)scratch_alloc(&idx_hdr, (size_t)pass_count * sizeof(int64_t)); - if (!match_idx) + if (!match_idx) { + scratch_free(fidx_hdr); return exec_filter_seq(input, pred, ncols, pass_count); + } - { + if (fidx_par) { + fidx.match_idx = match_idx; + ray_pool_dispatch(fidx_pool, fidx_fill_fn, &fidx, fidx_n_chunks); + } else { int64_t j = 0; ray_morsel_t mp; ray_morsel_init(&mp, pred); @@ -258,6 +341,7 @@ ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { row_base += mp.morsel_len; } } + scratch_free(fidx_hdr); /* Parallel gather — same pattern as sort gather */ ray_pool_t* pool = ray_pool_get(); @@ -511,6 +595,39 @@ ray_t* exec_filter_head(ray_t* input, ray_t* pred, int64_t limit) { * Reuses the same parallel multi-column gather as exec_filter. * ============================================================================ */ +/* Parallel per-seg match_idx fill for sel_compact: each worker writes + * its segments' passing rows at the precomputed disjoint offsets. */ +typedef struct { + const uint8_t* flags; + const uint32_t* offsets; + const uint16_t* idx; + const int64_t* seg_out; /* exclusive prefix of per-seg pass counts */ + int64_t* match_idx; + int64_t nrows; +} scidx_ctx_t; + +static void scidx_fill_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { + (void)wid; + scidx_ctx_t* c = (scidx_ctx_t*)ctxp; + for (int64_t seg = start; seg < end; seg++) { + uint8_t f = c->flags[seg]; + if (f == RAY_SEL_NONE) continue; + int64_t seg_start = seg * RAY_MORSEL_ELEMS; + int64_t seg_end = seg_start + RAY_MORSEL_ELEMS; + if (seg_end > c->nrows) seg_end = c->nrows; + int64_t j = c->seg_out[seg]; + if (f == RAY_SEL_ALL) { + for (int64_t r = seg_start; r < seg_end; r++) + c->match_idx[j++] = r; + } else { + const uint16_t* slice = c->idx + c->offsets[seg]; + uint32_t n = c->offsets[seg + 1] - c->offsets[seg]; + for (uint32_t i = 0; i < n; i++) + c->match_idx[j++] = seg_start + slice[i]; + } + } +} + /* keep-set membership: is column name-sym `nm` present in keep_syms? */ static inline bool sel_compact_keep(int64_t nm, const int64_t* keep_syms, int keep_n) { for (int j = 0; j < keep_n; j++) @@ -576,7 +693,10 @@ ray_t* sel_compact(ray_graph_t* g, ray_t* tbl, ray_t* sel, int64_t ncols = ray_table_ncols(tbl); if (ncols <= 0) { ray_retain(tbl); return tbl; } - /* Build match_idx from bitmap */ + /* Build match_idx from bitmap. Per-seg pass counts are derivable + * from the rowsel itself (ALL→seg_len, MIX→offsets delta, NONE→0), + * so this parallelizes as: tiny serial prefix over segs → parallel + * per-seg fill at disjoint offsets. */ ray_t* idx_hdr = NULL; int64_t* match_idx = (int64_t*)scratch_alloc(&idx_hdr, (size_t)pass_count * sizeof(int64_t)); @@ -587,21 +707,46 @@ ray_t* sel_compact(ray_graph_t* g, ray_t* tbl, ray_t* sel, const uint32_t* offsets = ray_rowsel_offsets(sel); const uint16_t* idx = ray_rowsel_idx(sel); uint32_t n_segs = meta->n_segs; - int64_t j = 0; - for (uint32_t seg = 0; seg < n_segs; seg++) { - uint8_t f = flags[seg]; - if (f == RAY_SEL_NONE) continue; - int64_t seg_start = (int64_t)seg * RAY_MORSEL_ELEMS; - int64_t seg_end = seg_start + RAY_MORSEL_ELEMS; - if (seg_end > nrows) seg_end = nrows; - if (f == RAY_SEL_ALL) { - for (int64_t r = seg_start; r < seg_end; r++) - match_idx[j++] = r; - } else { - const uint16_t* slice = idx + offsets[seg]; - uint32_t n = offsets[seg + 1] - offsets[seg]; - for (uint32_t i = 0; i < n; i++) - match_idx[j++] = seg_start + slice[i]; + ray_pool_t* sc_pool = ray_pool_get(); + ray_t* soff_hdr = NULL; + int64_t* seg_out = (sc_pool && nrows >= RAY_PARALLEL_THRESHOLD) + ? (int64_t*)scratch_alloc(&soff_hdr, (size_t)n_segs * sizeof(int64_t)) + : NULL; + if (seg_out) { + int64_t cum = 0; + for (uint32_t seg = 0; seg < n_segs; seg++) { + seg_out[seg] = cum; + uint8_t f = flags[seg]; + if (f == RAY_SEL_NONE) continue; + int64_t seg_start = (int64_t)seg * RAY_MORSEL_ELEMS; + int64_t seg_end = seg_start + RAY_MORSEL_ELEMS; + if (seg_end > nrows) seg_end = nrows; + cum += (f == RAY_SEL_ALL) ? (seg_end - seg_start) + : (int64_t)(offsets[seg + 1] - offsets[seg]); + } + scidx_ctx_t sc = { + .flags = flags, .offsets = offsets, .idx = idx, + .seg_out = seg_out, .match_idx = match_idx, .nrows = nrows, + }; + ray_pool_dispatch(sc_pool, scidx_fill_fn, &sc, (int64_t)n_segs); + scratch_free(soff_hdr); + } else { + int64_t j = 0; + for (uint32_t seg = 0; seg < n_segs; seg++) { + uint8_t f = flags[seg]; + if (f == RAY_SEL_NONE) continue; + int64_t seg_start = (int64_t)seg * RAY_MORSEL_ELEMS; + int64_t seg_end = seg_start + RAY_MORSEL_ELEMS; + if (seg_end > nrows) seg_end = nrows; + if (f == RAY_SEL_ALL) { + for (int64_t r = seg_start; r < seg_end; r++) + match_idx[j++] = r; + } else { + const uint16_t* slice = idx + offsets[seg]; + uint32_t n = offsets[seg + 1] - offsets[seg]; + for (uint32_t i = 0; i < n; i++) + match_idx[j++] = seg_start + slice[i]; + } } } } From 12a261876807bd9e6ba1b02a6046121b92430c6e Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Tue, 21 Jul 2026 18:52:43 +0300 Subject: [PATCH 7/8] perf: parallel where builtin, gather_by_idx, and chunk-task dispatch - ray_where_fn: 3-phase chunk compaction on the pool (was fully serial). - gather_by_idx: fixed-width value gathers dispatched over disjoint output ranges (null-bit propagation stays serial - shared-word bit writes would race). - exec_filter/where chunk phases now use ray_pool_dispatch_n (one task per chunk); ray_pool_dispatch morselizes total_elems by 1024, so passing chunk counts gave only ~2 tasks for 100M rows. 100M rows local c24: where 88->25ms, at-gather 80->31ms, 2-col where-select 138->20ms (6.8x). make test 3635/3635. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/lang/eval.c | 64 +++++++++++++++++++++++++----------- src/ops/builtins.c | 82 ++++++++++++++++++++++++++++++++++++++++++++++ src/ops/filter.c | 23 ++++++++----- 3 files changed, 141 insertions(+), 28 deletions(-) diff --git a/src/lang/eval.c b/src/lang/eval.c index 53c5d52b..2f32b159 100644 --- a/src/lang/eval.c +++ b/src/lang/eval.c @@ -1432,6 +1432,32 @@ ray_t* call_fn2(ray_t* fn, ray_t* a, ray_t* b) { * Sorting builtins * ══════════════════════════════════════════ */ +/* Parallel fixed-width index gather for gather_by_idx: value memcpy loops + * split across the worker pool. Null-bit propagation stays serial (bit + * writes within a shared word would race across range boundaries). */ +typedef struct { + const char* src; + char* dst; + const int64_t* idx; + uint8_t esz; +} gbi_ctx_t; + +static void gbi_fill_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { + (void)wid; + gbi_ctx_t* c = (gbi_ctx_t*)ctxp; + const char* src = c->src; + char* dst = c->dst; + const int64_t* idx = c->idx; + switch (c->esz) { + case 8: for (int64_t i = start; i < end; i++) memcpy(dst + i*8, src + idx[i]*8, 8); break; + case 4: for (int64_t i = start; i < end; i++) memcpy(dst + i*4, src + idx[i]*4, 4); break; + case 2: for (int64_t i = start; i < end; i++) memcpy(dst + i*2, src + idx[i]*2, 2); break; + case 1: for (int64_t i = start; i < end; i++) dst[i] = src[idx[i]]; break; + case 16: for (int64_t i = start; i < end; i++) memcpy(dst + i*16, src + idx[i]*16, 16); break; + default: for (int64_t i = start; i < end; i++) memcpy(dst + i*c->esz, src + idx[i]*c->esz, c->esz); break; + } +} + /* Reorder vector elements by an index array */ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) { int8_t type = vec->type; @@ -1465,15 +1491,14 @@ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) { if (RAY_IS_ERR(result)) return result; result->len = n; uint8_t esz = (uint8_t)RAY_SYM_ELEM(w); - char* src = (char*)ray_data(vec); - char* dst = (char*)ray_data(result); - switch (esz) { - case 8: for (int64_t i = 0; i < n; i++) memcpy(dst + i*8, src + idx[i]*8, 8); break; - case 4: for (int64_t i = 0; i < n; i++) memcpy(dst + i*4, src + idx[i]*4, 4); break; - case 2: for (int64_t i = 0; i < n; i++) memcpy(dst + i*2, src + idx[i]*2, 2); break; - case 1: for (int64_t i = 0; i < n; i++) dst[i] = src[idx[i]]; break; - default: for (int64_t i = 0; i < n; i++) memcpy(dst + i*esz, src + idx[i]*esz, esz); break; - } + gbi_ctx_t gc = { .src = (const char*)ray_data(vec), + .dst = (char*)ray_data(result), + .idx = idx, .esz = esz }; + ray_pool_t* gpool = ray_pool_get(); + if (gpool && n >= RAY_PARALLEL_THRESHOLD) + ray_pool_dispatch(gpool, gbi_fill_fn, &gc, n); + else + gbi_fill_fn(&gc, 0, 0, n); if (has_nulls) { for (int64_t i = 0; i < n; i++) if (ray_vec_is_null(vec, idx[i])) @@ -1506,17 +1531,16 @@ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) { if (RAY_IS_ERR(result)) return result; result->len = n; uint8_t esz = ray_type_sizes[type]; - char* src = (char*)ray_data(vec); - char* dst = (char*)ray_data(result); - /* Typed gather — compiler constant esz enables vectorization, alias-safe */ - switch (esz) { - case 8: for (int64_t i = 0; i < n; i++) memcpy(dst + i*8, src + idx[i]*8, 8); break; - case 4: for (int64_t i = 0; i < n; i++) memcpy(dst + i*4, src + idx[i]*4, 4); break; - case 2: for (int64_t i = 0; i < n; i++) memcpy(dst + i*2, src + idx[i]*2, 2); break; - case 1: for (int64_t i = 0; i < n; i++) dst[i] = src[idx[i]]; break; - default: for (int64_t i = 0; i < n; i++) memcpy(dst + i*esz, src + idx[i]*esz, esz); break; - case 16: for (int64_t i = 0; i < n; i++) memcpy(dst + i*16, src + idx[i]*16, 16); break; - } + /* Typed gather — compiler constant esz enables vectorization, alias-safe; + * pool-parallel over disjoint output ranges for large gathers. */ + gbi_ctx_t gc = { .src = (const char*)ray_data(vec), + .dst = (char*)ray_data(result), + .idx = idx, .esz = esz }; + ray_pool_t* gpool = ray_pool_get(); + if (gpool && n >= RAY_PARALLEL_THRESHOLD) + ray_pool_dispatch(gpool, gbi_fill_fn, &gc, n); + else + gbi_fill_fn(&gc, 0, 0, n); /* Propagate null bitmap */ if (has_nulls) { diff --git a/src/ops/builtins.c b/src/ops/builtins.c index 6b631f09..7ca5d638 100644 --- a/src/ops/builtins.c +++ b/src/ops/builtins.c @@ -2100,11 +2100,93 @@ ray_t* ray_nil_fn(ray_t* x) { } /* (where bool-vec) -> indices of true values */ + +/* Parallel where: 3-phase chunk compaction (parallel per-chunk popcount → + * tiny serial prefix → parallel per-chunk emit at disjoint offsets). */ +#define WHERE_CHUNK_ROWS 65536 + +typedef struct { + const uint8_t* data; + int64_t n; + int64_t chunk_rows; + int64_t* chunk_counts; /* popcount, then exclusive-prefix offsets */ + int64_t* out; /* emit-phase output (NULL during count) */ +} where_ctx_t; + +/* Dispatched via ray_pool_dispatch_n — each task is ONE chunk ([i,i+1)). */ +static void where_count_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { + (void)wid; + where_ctx_t* c = (where_ctx_t*)ctxp; + for (int64_t ch = start; ch < end; ch++) { + int64_t lo = ch * c->chunk_rows; + int64_t hi = lo + c->chunk_rows; + if (hi > c->n) hi = c->n; + int64_t cnt = 0; + for (int64_t i = lo; i < hi; i++) cnt += (c->data[i] != 0); + c->chunk_counts[ch] = cnt; + } +} + +/* Word-at-a-time zero skip: sparse masks — the common WHERE shape — + * run at memory speed, falling to per-byte only inside a live word. */ +static void where_emit_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { + (void)wid; + where_ctx_t* c = (where_ctx_t*)ctxp; + for (int64_t ch = start; ch < end; ch++) { + int64_t lo = ch * c->chunk_rows; + int64_t hi = lo + c->chunk_rows; + if (hi > c->n) hi = c->n; + int64_t j = c->chunk_counts[ch]; + int64_t i = lo; + int64_t h8 = lo + ((hi - lo) & ~7LL); + for (; i < h8; i += 8) { + uint64_t w; + memcpy(&w, c->data + i, 8); + if (!w) continue; + for (int64_t k = 0; k < 8; k++) + if (c->data[i + k]) c->out[j++] = i + k; + } + for (; i < hi; i++) + if (c->data[i]) c->out[j++] = i; + } +} + ray_t* ray_where_fn(ray_t* x) { if (!ray_is_vec(x) || x->type != RAY_BOOL) return ray_error("type", "where: argument must be a b8 vector, got %s", ray_type_name(x->type)); const uint8_t* data = (const uint8_t*)ray_data(x); int64_t n = x->len; + + ray_pool_t* pool = ray_pool_get(); + if (pool && n >= RAY_PARALLEL_THRESHOLD) { + /* One task per chunk (ray_pool_dispatch_n); chunk grows for inputs + * that would exceed the pool's task cap. */ + int64_t chunk_rows = WHERE_CHUNK_ROWS; + while ((n + chunk_rows - 1) / chunk_rows > 60000) chunk_rows *= 2; + int64_t n_chunks = (n + chunk_rows - 1) / chunk_rows; + ray_t* cc_hdr = ray_alloc((size_t)n_chunks * sizeof(int64_t)); + int64_t* chunk_counts = cc_hdr ? (int64_t*)ray_data(cc_hdr) : NULL; + if (chunk_counts) { + where_ctx_t wc = { .data = data, .n = n, .chunk_rows = chunk_rows, + .chunk_counts = chunk_counts, .out = NULL }; + ray_pool_dispatch_n(pool, where_count_fn, &wc, (uint32_t)n_chunks); + int64_t cum = 0; + for (int64_t ch = 0; ch < n_chunks; ch++) { + int64_t v = chunk_counts[ch]; + chunk_counts[ch] = cum; + cum += v; + } + ray_t* result = ray_vec_new(RAY_I64, cum); + if (RAY_IS_ERR(result)) { ray_release(cc_hdr); return result; } + wc.out = (int64_t*)ray_data(result); + ray_pool_dispatch_n(pool, where_emit_fn, &wc, (uint32_t)n_chunks); + result->len = cum; + ray_release(cc_hdr); + return result; + } + /* scratch OOM → sequential fallback below */ + } + /* Count pass: `cnt += byte != 0` vectorizes; the branchy form costs * a mispredict-prone compare per element. */ int64_t cnt = 0; diff --git a/src/ops/filter.c b/src/ops/filter.c index 313fb534..3ecc2ff5 100644 --- a/src/ops/filter.c +++ b/src/ops/filter.c @@ -215,16 +215,18 @@ static ray_t* exec_filter_seq(ray_t* input, ray_t* pred, int64_t ncols, typedef struct { const uint8_t* bits; int64_t nrows; + int64_t chunk_rows; int64_t* chunk_counts; /* [n_chunks]: popcount, then prefix offset */ int64_t* match_idx; /* fill-phase output (NULL during count) */ } fidx_ctx_t; +/* Dispatched via ray_pool_dispatch_n — each task is ONE chunk ([i,i+1)). */ static void fidx_count_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { (void)wid; fidx_ctx_t* c = (fidx_ctx_t*)ctxp; for (int64_t ch = start; ch < end; ch++) { - int64_t lo = ch * FIDX_CHUNK_ROWS; - int64_t hi = lo + FIDX_CHUNK_ROWS; + int64_t lo = ch * c->chunk_rows; + int64_t hi = lo + c->chunk_rows; if (hi > c->nrows) hi = c->nrows; int64_t n = 0; for (int64_t i = lo; i < hi; i++) n += c->bits[i] ? 1 : 0; @@ -236,8 +238,8 @@ static void fidx_fill_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { (void)wid; fidx_ctx_t* c = (fidx_ctx_t*)ctxp; for (int64_t ch = start; ch < end; ch++) { - int64_t lo = ch * FIDX_CHUNK_ROWS; - int64_t hi = lo + FIDX_CHUNK_ROWS; + int64_t lo = ch * c->chunk_rows; + int64_t hi = lo + c->chunk_rows; if (hi > c->nrows) hi = c->nrows; int64_t j = c->chunk_counts[ch]; /* prefix offset after phase 2 */ for (int64_t i = lo; i < hi; i++) @@ -258,7 +260,7 @@ ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { bool fidx_par = fidx_pool && pred->type == RAY_BOOL && !ray_is_lazy(pred) && pred->len >= RAY_PARALLEL_THRESHOLD; ray_t* fidx_hdr = NULL; - fidx_ctx_t fidx = { .bits = NULL, .nrows = 0, + fidx_ctx_t fidx = { .bits = NULL, .nrows = 0, .chunk_rows = FIDX_CHUNK_ROWS, .chunk_counts = NULL, .match_idx = NULL }; int64_t fidx_n_chunks = 0; @@ -267,11 +269,15 @@ ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { if (fidx_par) { fidx.bits = (const uint8_t*)ray_data(pred); fidx.nrows = pred->len; - fidx_n_chunks = (pred->len + FIDX_CHUNK_ROWS - 1) / FIDX_CHUNK_ROWS; + /* One task per chunk (dispatch_n); chunk grows past the task cap. */ + while ((fidx.nrows + fidx.chunk_rows - 1) / fidx.chunk_rows > 60000) + fidx.chunk_rows *= 2; + fidx_n_chunks = (fidx.nrows + fidx.chunk_rows - 1) / fidx.chunk_rows; fidx.chunk_counts = (int64_t*)scratch_alloc(&fidx_hdr, (size_t)fidx_n_chunks * sizeof(int64_t)); if (fidx.chunk_counts) - ray_pool_dispatch(fidx_pool, fidx_count_fn, &fidx, fidx_n_chunks); + ray_pool_dispatch_n(fidx_pool, fidx_count_fn, &fidx, + (uint32_t)fidx_n_chunks); else fidx_par = false; } @@ -328,7 +334,8 @@ ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { if (fidx_par) { fidx.match_idx = match_idx; - ray_pool_dispatch(fidx_pool, fidx_fill_fn, &fidx, fidx_n_chunks); + ray_pool_dispatch_n(fidx_pool, fidx_fill_fn, &fidx, + (uint32_t)fidx_n_chunks); } else { int64_t j = 0; ray_morsel_t mp; From 5aa7b641fc1d023a7a5f8251bc6b967b3da6941d Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Wed, 22 Jul 2026 15:21:22 +0300 Subject: [PATCH 8/8] fix(review): harden parallel paths per skeptic review Blockers (DA binary-agg y-column): - eligibility now requires a plain numeric/temporal y; nullable integer/temporal y stays on the HT path (da_accum_row's pair branch has no y-side sentinel machinery - nulls would accumulate as values) - an FP y with HAS_NULLS sets da_any_nullable so nn[] is allocated and the emitter divides by the non-null PAIR count, not the group count Majors: - all new parallel gates require pool->n_workers > 0 (a -c 1 pool exists with 0 workers; ring fill + atomics + rc_sync were pure overhead, and the OP_IF eager reroute lost to the selected path serially - the Q22/Q25 c1 regression) - chunked dispatch_n call sites cap chunks at 1024 = the pool's initial ring capacity, so the ring never grows (dispatch_n clamps and silently DROPS tasks if ring growth fails -> uninitialized prefix entries -> OOB writes) - sel_compact seg fill switched to dispatch_n over seg-chunks (ray_pool_dispatch over segs gave 1 task under 8.4M rows) - gather_by_idx parallel path guarded by ray_parallel_flag == 0 (leaf utility, 35+ call sites; nested dispatch would corrupt the single-producer task ring) Nits: stray time.h include, restored v2-gate comment, RAY_PARALLEL_THRESHOLD symbol in pivot.c. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/lang/eval.c | 19 +++++++++++--- src/ops/builtins.c | 14 +++++++--- src/ops/filter.c | 65 +++++++++++++++++++++++++++++++--------------- src/ops/group.c | 35 ++++++++++++++++++++----- src/ops/pivot.c | 21 ++++++++++----- 5 files changed, 113 insertions(+), 41 deletions(-) diff --git a/src/lang/eval.c b/src/lang/eval.c index 2f32b159..a609c5a3 100644 --- a/src/lang/eval.c +++ b/src/lang/eval.c @@ -1434,7 +1434,20 @@ ray_t* call_fn2(ray_t* fn, ray_t* a, ray_t* b) { /* Parallel fixed-width index gather for gather_by_idx: value memcpy loops * split across the worker pool. Null-bit propagation stays serial (bit - * writes within a shared word would race across range boundaries). */ + * writes within a shared word would race across range boundaries). + * + * gather_by_idx is a LEAF utility with 35+ call sites; unlike op-level + * dispatch points it cannot know its execution context, so the parallel + * path additionally requires no dispatch in flight (ray_parallel_flag) — + * ray_pool_dispatch is single-producer, and a nested dispatch from a + * worker would corrupt the task ring. */ +extern _Atomic(uint32_t) ray_parallel_flag; /* core/platform.h */ + +static inline bool gbi_par_ok(ray_pool_t* p, int64_t n) { + return p && p->n_workers > 0 && n >= RAY_PARALLEL_THRESHOLD && + atomic_load_explicit(&ray_parallel_flag, memory_order_relaxed) == 0; +} + typedef struct { const char* src; char* dst; @@ -1495,7 +1508,7 @@ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) { .dst = (char*)ray_data(result), .idx = idx, .esz = esz }; ray_pool_t* gpool = ray_pool_get(); - if (gpool && n >= RAY_PARALLEL_THRESHOLD) + if (gbi_par_ok(gpool, n)) ray_pool_dispatch(gpool, gbi_fill_fn, &gc, n); else gbi_fill_fn(&gc, 0, 0, n); @@ -1537,7 +1550,7 @@ ray_t* gather_by_idx(ray_t* vec, int64_t* idx, int64_t n) { .dst = (char*)ray_data(result), .idx = idx, .esz = esz }; ray_pool_t* gpool = ray_pool_get(); - if (gpool && n >= RAY_PARALLEL_THRESHOLD) + if (gbi_par_ok(gpool, n)) ray_pool_dispatch(gpool, gbi_fill_fn, &gc, n); else gbi_fill_fn(&gc, 0, 0, n); diff --git a/src/ops/builtins.c b/src/ops/builtins.c index 7ca5d638..7685caea 100644 --- a/src/ops/builtins.c +++ b/src/ops/builtins.c @@ -2158,12 +2158,18 @@ ray_t* ray_where_fn(ray_t* x) { int64_t n = x->len; ray_pool_t* pool = ray_pool_get(); - if (pool && n >= RAY_PARALLEL_THRESHOLD) { - /* One task per chunk (ray_pool_dispatch_n); chunk grows for inputs - * that would exceed the pool's task cap. */ + if (pool && pool->n_workers > 0 && n >= RAY_PARALLEL_THRESHOLD) { + /* One task per chunk (ray_pool_dispatch_n). Chunk count is capped + * at 1024 — the pool's INITIAL ring capacity — so dispatch_n never + * needs to grow the ring (its clamp-on-grow-failure silently DROPS + * tasks, which here would mean uninitialized prefix entries and + * out-of-bounds emit offsets). 1024 tasks is ample granularity. */ int64_t chunk_rows = WHERE_CHUNK_ROWS; - while ((n + chunk_rows - 1) / chunk_rows > 60000) chunk_rows *= 2; + while ((n + chunk_rows - 1) / chunk_rows > 1024) chunk_rows *= 2; int64_t n_chunks = (n + chunk_rows - 1) / chunk_rows; + /* ops/internal.h's scratch_alloc idiom is unavailable here (that + * header clashes with builtins.c-local helpers); ray_alloc + + * ray_release below is its exact expansion. */ ray_t* cc_hdr = ray_alloc((size_t)n_chunks * sizeof(int64_t)); int64_t* chunk_counts = cc_hdr ? (int64_t*)ray_data(cc_hdr) : NULL; if (chunk_counts) { diff --git a/src/ops/filter.c b/src/ops/filter.c index 3ecc2ff5..f876aec6 100644 --- a/src/ops/filter.c +++ b/src/ops/filter.c @@ -257,7 +257,8 @@ ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { * parallel chunked count/fill; anything else (lazy / morsel-backed * pred, small input) keeps the sequential morsel sweep. */ ray_pool_t* fidx_pool = ray_pool_get(); - bool fidx_par = fidx_pool && pred->type == RAY_BOOL && !ray_is_lazy(pred) && + bool fidx_par = fidx_pool && fidx_pool->n_workers > 0 && + pred->type == RAY_BOOL && !ray_is_lazy(pred) && pred->len >= RAY_PARALLEL_THRESHOLD; ray_t* fidx_hdr = NULL; fidx_ctx_t fidx = { .bits = NULL, .nrows = 0, .chunk_rows = FIDX_CHUNK_ROWS, @@ -269,8 +270,12 @@ ray_t* exec_filter(ray_graph_t* g, ray_op_t* op, ray_t* input, ray_t* pred) { if (fidx_par) { fidx.bits = (const uint8_t*)ray_data(pred); fidx.nrows = pred->len; - /* One task per chunk (dispatch_n); chunk grows past the task cap. */ - while ((fidx.nrows + fidx.chunk_rows - 1) / fidx.chunk_rows > 60000) + /* One task per chunk (dispatch_n). Chunk count is capped at 1024 — + * the pool's INITIAL ring capacity — so dispatch_n never needs to + * grow the ring (its clamp-on-grow-failure silently DROPS tasks, + * which here would leave uninitialized prefix entries and cause + * out-of-bounds fill offsets). 1024 tasks is ample granularity. */ + while ((fidx.nrows + fidx.chunk_rows - 1) / fidx.chunk_rows > 1024) fidx.chunk_rows *= 2; fidx_n_chunks = (fidx.nrows + fidx.chunk_rows - 1) / fidx.chunk_rows; fidx.chunk_counts = (int64_t*)scratch_alloc(&fidx_hdr, @@ -603,7 +608,9 @@ ray_t* exec_filter_head(ray_t* input, ray_t* pred, int64_t limit) { * ============================================================================ */ /* Parallel per-seg match_idx fill for sel_compact: each worker writes - * its segments' passing rows at the precomputed disjoint offsets. */ + * its segments' passing rows at the precomputed disjoint offsets. + * Dispatched via ray_pool_dispatch_n — one task per SEG-CHUNK ([i,i+1)), + * segs_per_chunk segments each. */ typedef struct { const uint8_t* flags; const uint32_t* offsets; @@ -611,26 +618,33 @@ typedef struct { const int64_t* seg_out; /* exclusive prefix of per-seg pass counts */ int64_t* match_idx; int64_t nrows; + uint32_t n_segs; + uint32_t segs_per_chunk; } scidx_ctx_t; static void scidx_fill_fn(void* ctxp, uint32_t wid, int64_t start, int64_t end) { (void)wid; scidx_ctx_t* c = (scidx_ctx_t*)ctxp; - for (int64_t seg = start; seg < end; seg++) { - uint8_t f = c->flags[seg]; - if (f == RAY_SEL_NONE) continue; - int64_t seg_start = seg * RAY_MORSEL_ELEMS; - int64_t seg_end = seg_start + RAY_MORSEL_ELEMS; - if (seg_end > c->nrows) seg_end = c->nrows; - int64_t j = c->seg_out[seg]; - if (f == RAY_SEL_ALL) { - for (int64_t r = seg_start; r < seg_end; r++) - c->match_idx[j++] = r; - } else { - const uint16_t* slice = c->idx + c->offsets[seg]; - uint32_t n = c->offsets[seg + 1] - c->offsets[seg]; - for (uint32_t i = 0; i < n; i++) - c->match_idx[j++] = seg_start + slice[i]; + for (int64_t ch = start; ch < end; ch++) { + uint32_t seg0 = (uint32_t)ch * c->segs_per_chunk; + uint32_t seg1 = seg0 + c->segs_per_chunk; + if (seg1 > c->n_segs) seg1 = c->n_segs; + for (uint32_t seg = seg0; seg < seg1; seg++) { + uint8_t f = c->flags[seg]; + if (f == RAY_SEL_NONE) continue; + int64_t seg_start = (int64_t)seg * RAY_MORSEL_ELEMS; + int64_t seg_end = seg_start + RAY_MORSEL_ELEMS; + if (seg_end > c->nrows) seg_end = c->nrows; + int64_t j = c->seg_out[seg]; + if (f == RAY_SEL_ALL) { + for (int64_t r = seg_start; r < seg_end; r++) + c->match_idx[j++] = r; + } else { + const uint16_t* slice = c->idx + c->offsets[seg]; + uint32_t n = c->offsets[seg + 1] - c->offsets[seg]; + for (uint32_t i = 0; i < n; i++) + c->match_idx[j++] = seg_start + slice[i]; + } } } } @@ -716,7 +730,8 @@ ray_t* sel_compact(ray_graph_t* g, ray_t* tbl, ray_t* sel, uint32_t n_segs = meta->n_segs; ray_pool_t* sc_pool = ray_pool_get(); ray_t* soff_hdr = NULL; - int64_t* seg_out = (sc_pool && nrows >= RAY_PARALLEL_THRESHOLD) + int64_t* seg_out = (sc_pool && sc_pool->n_workers > 0 && + nrows >= RAY_PARALLEL_THRESHOLD) ? (int64_t*)scratch_alloc(&soff_hdr, (size_t)n_segs * sizeof(int64_t)) : NULL; if (seg_out) { @@ -731,11 +746,19 @@ ray_t* sel_compact(ray_graph_t* g, ray_t* tbl, ray_t* sel, cum += (f == RAY_SEL_ALL) ? (seg_end - seg_start) : (int64_t)(offsets[seg + 1] - offsets[seg]); } + /* One task per seg-chunk (dispatch_n; ray_pool_dispatch would + * morselize the seg RANGE by 8192 segs → 1 task under 8.4M + * rows). Chunk count capped at 1024 = the pool's initial ring + * capacity, so the ring never needs to grow (see exec_filter). */ + uint32_t spc = 64; + while ((n_segs + spc - 1) / spc > 1024) spc *= 2; + uint32_t sc_chunks = (n_segs + spc - 1) / spc; scidx_ctx_t sc = { .flags = flags, .offsets = offsets, .idx = idx, .seg_out = seg_out, .match_idx = match_idx, .nrows = nrows, + .n_segs = n_segs, .segs_per_chunk = spc, }; - ray_pool_dispatch(sc_pool, scidx_fill_fn, &sc, (int64_t)n_segs); + ray_pool_dispatch_n(sc_pool, scidx_fill_fn, &sc, sc_chunks); scratch_free(soff_hdr); } else { int64_t j = 0; diff --git a/src/ops/group.c b/src/ops/group.c index abb3100a..dbe81daa 100644 --- a/src/ops/group.c +++ b/src/ops/group.c @@ -34,7 +34,6 @@ #include "core/runtime.h" /* __VM — per-thread group-key cardinality hint */ #include -#include /* Group accumulators use F64 lanes for both floating storage widths. Keep the * source read width explicit so F32 payload bits are never mistaken for an @@ -8604,6 +8603,8 @@ static ray_t* exec_group_run(ray_graph_t* g, ray_op_t* op, ray_t* tbl, ray_op_ext_t* ext = find_ext(g, op->id); if (!ext) return ray_error("nyi", NULL); + /* v2 doesn't implement the top-count emit filter (old-engine feature); + * when one is active, stay on the legacy path that honors it. */ if (ray_agg_engine_v2 && group_limit == 0 && !ray_group_emit_filter_get().enabled && agg_v2_can_handle(g, op, tbl)) @@ -9594,11 +9595,26 @@ da_path:; for (uint32_t a = 0; a < n_aggs && da_eligible; a++) { uint16_t aop = ext->agg_ops[a]; /* Binary aggregators (PEARSON/COV/SCOV/WSUM/WAVG) are handled by - * the DA co-moment slots (sum_y/sumsq_y/sumxy); they only need the - * paired y-column present. (Previously forced to the slow HT - * scatter path — the cause of poor multi-thread scaling.) */ + * the DA co-moment slots (sum_y/sumsq_y/sumxy). (Previously + * forced to the slow HT scatter path — the cause of poor + * multi-thread scaling.) Eligibility needs a paired y-column of + * a plain numeric/temporal type; da_accum_row's pair branch has + * NO sentinel machinery for the y side (only the FP NaN check), + * so a nullable integer/temporal y — whose null sentinel would + * be accumulated as a real value — must stay on the HT path. */ if (agg_is_binary_agg(aop)) { - if (!agg_vecs2 || !agg_vecs2[a]) { da_eligible = false; } + ray_t* yv2 = (agg_vecs2 ? agg_vecs2[a] : NULL); + if (!yv2) { da_eligible = false; } + else { + int8_t t2 = yv2->type; + bool y_fp = group_fp_type(t2); + bool y_int = (t2 == RAY_I16 || t2 == RAY_I32 || t2 == RAY_I64 || + t2 == RAY_DATE || t2 == RAY_TIME || t2 == RAY_TIMESTAMP || + t2 == RAY_BOOL || t2 == RAY_U8); + if (!y_fp && !y_int) da_eligible = false; /* SYM/STR/GUID/… */ + else if (!y_fp && (yv2->attrs & RAY_ATTR_HAS_NULLS)) + da_eligible = false; /* nullable int y */ + } } if (aop == OP_MEDIAN) da_eligible = false; if (aop == OP_QUANTILE) da_eligible = false; @@ -9780,12 +9796,19 @@ da_path:; * with HAS_NULLS use sentinel-skip. */ bool da_any_nullable = false; for (uint32_t a = 0; a < n_aggs; a++) { - /* y-side plumbing for binary aggs (PEARSON/COV/WAVG/WSUM). */ + /* y-side plumbing for binary aggs (PEARSON/COV/WAVG/WSUM). + * Eligibility (above) already rejected nullable-int and + * non-numeric y; an FP y may still carry NaNs — the accum + * branch skips those PAIRS, so nn must be allocated or the + * emitter would divide by the full group count. */ agg_ptrs2[a] = NULL; agg_types2[a] = 0; if (agg_is_binary_agg(ext->agg_ops[a]) && agg_vecs2 && agg_vecs2[a]) { agg_ptrs2[a] = ray_data(agg_vecs2[a]); agg_types2[a] = agg_vecs2[a]->type; agg_pair_mask |= ((uint64_t)1 << a); + if (group_fp_type(agg_vecs2[a]->type) && + (agg_vecs2[a]->attrs & RAY_ATTR_HAS_NULLS)) + da_any_nullable = true; } if (agg_prod[a].enabled) { /* Fused product: F64 accumulate, no source vec. */ diff --git a/src/ops/pivot.c b/src/ops/pivot.c index edef5b83..08e2ec13 100644 --- a/src/ops/pivot.c +++ b/src/ops/pivot.c @@ -488,12 +488,19 @@ static ray_t* exec_if_selected(ray_graph_t* g, ray_op_t* op, ray_t* cond_v) { if (!then_op || !else_op) return NULL; /* Both branches trivial AND type-safe for the eager fill → eager - * (parallel elementwise) wins over the serial id-scaffolding here. */ - if (op->out_type != RAY_STR && - if_branch_trivial(g, then_op) && if_branch_trivial(g, else_op) && - if_type_eager_ok(then_op->out_type, op->out_type) && - if_type_eager_ok(else_op->out_type, op->out_type)) - return NULL; + * (POOL-PARALLEL elementwise) wins over the serial id-scaffolding + * here. Only when workers exist: with a 0-worker pool (-c 1) the + * eager fill runs serially over ALL rows for BOTH branches, which + * loses to the selected path's touch-only-passing-rows scheme. */ + { + ray_pool_t* rp = ray_pool_get(); + if (rp && rp->n_workers > 0 && + op->out_type != RAY_STR && + if_branch_trivial(g, then_op) && if_branch_trivial(g, else_op) && + if_type_eager_ok(then_op->out_type, op->out_type) && + if_type_eager_ok(else_op->out_type, op->out_type)) + return NULL; + } if_branch_plan_t then_plan, else_plan; if (!if_make_branch_plan(g, then_op, &then_plan) || @@ -809,7 +816,7 @@ static ray_t* exec_if_eager(ray_graph_t* g, ray_op_t* op) { } ray_pool_t* pool = ray_pool_get(); - bool par = pool && len >= 65536; + bool par = pool && pool->n_workers > 0 && len >= RAY_PARALLEL_THRESHOLD; if (par && out_type == RAY_SYM) { /* Vector STR sides intern per element — serial only. Non-STR * vector sides must be SYM columns; warm each non-runtime