From 63d91b1ef5cee7b488184c7dfe9c4ba8fba69329 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Mon, 21 Sep 2026 10:56:19 +0300 Subject: [PATCH 1/3] fix(exec): release the input table of window, sort and limit unconditionally MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every node the executor evaluates returns an owned reference — the constant table node included, it retains its literal. The OP_WINDOW, OP_SORT and OP_HEAD cases released their input only when it differed from the graph's table, taking an equal pointer for a borrowed one. A query's root is a constant node over that very table, so the reference was never released: one input table per windowed, sorted or limited query. Invisible while a global kept the table alive; a whole table per call when the input was built for that call, as a service that windowed a freshly concatenated buffer on a timer found (#602). The four cases now release the input on every path, as the join and the plain head/tail cases already did. Test: window, sorted and limited selects over a table built per call, measured with the two-window bytes-allocated method of the other memory probes, plus a shared input reused across fifty calls. Co-Authored-By: Claude Fable 5.1 --- src/ops/exec.c | 27 ++++++++++-------- test/rfl/mem/query_input_release.rfl | 42 ++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 11 deletions(-) create mode 100644 test/rfl/mem/query_input_release.rfl diff --git a/src/ops/exec.c b/src/ops/exec.c index 93b549448..d762388dc 100644 --- a/src/ops/exec.c +++ b/src/ops/exec.c @@ -2610,20 +2610,26 @@ static ray_t* exec_node_inner(ray_graph_t* g, ray_op_t* op) { } case OP_SORT: { + /* exec_node hands back an OWNED reference for every node, + * the constant table node included (it retains its literal). + * A `!= g->table` guard here treated a child that evaluates + * to the query table as borrowed and leaked one table per + * call — the ordinary shape, since a select's root is a + * constant node over that very table. */ ray_t* input = exec_node(g, op_child(g, op, 0)); if (!input || RAY_IS_ERR(input)) return input; ray_t* tbl = (input->type == RAY_TABLE) ? input : g->table; /* Compact lazy selection before sort (needs dense data) */ if (g->selection && tbl && !RAY_IS_ERR(tbl) && tbl->type == RAY_TABLE) { ray_t* compacted = sel_compact(g, tbl, g->selection, NULL, 0); - if (input != g->table) ray_release(input); + ray_release(input); ray_release(g->selection); g->selection = NULL; input = compacted; tbl = compacted; } ray_t* result = exec_sort(g, op, tbl, 0); - if (input != g->table) ray_release(input); + ray_release(input); return result; } @@ -2835,20 +2841,21 @@ static ray_t* exec_node_inner(ray_graph_t* g, ray_op_t* op) { } case OP_WINDOW: { + /* Owned input, released unconditionally: see OP_SORT. */ ray_t* input = exec_node(g, op_child(g, op, 0)); if (!input || RAY_IS_ERR(input)) return input; ray_t* wdf = (input->type == RAY_TABLE) ? input : g->table; /* Compact lazy selection before window (needs dense data) */ if (g->selection && wdf && !RAY_IS_ERR(wdf) && wdf->type == RAY_TABLE) { ray_t* compacted = sel_compact(g, wdf, g->selection, NULL, 0); - if (input != g->table) ray_release(input); + ray_release(input); ray_release(g->selection); g->selection = NULL; input = compacted; wdf = compacted; } ray_t* result = exec_window(g, op, wdf); - if (input != g->table) ray_release(input); + ray_release(input); return result; } @@ -2865,14 +2872,14 @@ static ray_t* exec_node_inner(ray_graph_t* g, ray_op_t* op) { /* Compact lazy selection before sort */ if (g->selection && tbl && !RAY_IS_ERR(tbl) && tbl->type == RAY_TABLE) { ray_t* compacted = sel_compact(g, tbl, g->selection, NULL, 0); - if (sort_input != g->table) ray_release(sort_input); + ray_release(sort_input); /* owned: see OP_SORT */ ray_release(g->selection); g->selection = NULL; sort_input = compacted; tbl = compacted; } ray_t* result = exec_sort(g, child_op, tbl, n); - if (sort_input != g->table) ray_release(sort_input); + ray_release(sort_input); /* Top-level statement GC catches intermediates. */ return result; } @@ -2935,7 +2942,7 @@ static ray_t* exec_node_inner(ray_graph_t* g, ray_op_t* op) { ? filter_input : g->table; if (g->selection && ftbl && ftbl->type == RAY_TABLE) { ray_t* compacted = sel_compact(g, ftbl, g->selection, NULL, 0); - if (filter_input != g->table) ray_release(filter_input); + ray_release(filter_input); /* owned: see OP_SORT */ ray_release(g->selection); g->selection = NULL; filter_input = compacted; @@ -2949,15 +2956,13 @@ static ray_t* exec_node_inner(ray_graph_t* g, ray_op_t* op) { g->table = saved_table; if (!pred || RAY_IS_ERR(pred)) { - if (filter_input != saved_table) - ray_release(filter_input); + ray_release(filter_input); return pred; } ray_t* result = exec_filter_head(ftbl, pred, n); ray_release(pred); - if (filter_input != saved_table) - ray_release(filter_input); + ray_release(filter_input); /* Top-level statement GC catches intermediates. */ return result; } else { diff --git a/test/rfl/mem/query_input_release.rfl b/test/rfl/mem/query_input_release.rfl new file mode 100644 index 000000000..80d4c0c0c --- /dev/null +++ b/test/rfl/mem/query_input_release.rfl @@ -0,0 +1,42 @@ +;; query_input_release.rfl — a query over a table built for that one call +;; must give the table back. +;; +;; Every node the executor evaluates hands back an OWNED reference, the +;; constant table node included. The window, sort and limit cases once +;; skipped the release when the input equalled the graph's table, taking it +;; for a borrowed reference — and a query's root is a constant node over that +;; very table, so one input table leaked per call. Invisible while a global +;; kept the table alive; a whole table per call on a fresh one. +;; +;; Same method as the other probes in this directory: warm up, then two equal +;; windows of calls; the first may settle a slab or an arena, the second must +;; not move. `bytes-allocated` is exact, so the bound is a few KB; the table +;; below is ~19 KB, so a single leaked call already fails it. +(set syms (as 'SYMBOL (map (fn [i] (format "inst%" i)) (til 700)))) +(set mk (fn [p] (table [instrument rid v] (list (take syms 786) (as 'I64 (+ p (til 786))) (as 'I64 (til 786)))))) +(set heapb (fn [] (at (.sys.mem) 'bytes-allocated))) +(set probe (fn [f warm reps] (do (map f (til warm)) (map f (til reps)) (set h0 (heapb)) (map f (til reps)) (< (- (heapb) h0) 4096)))) + +;; partitioned + ordered, trailing frame (the shape of the report) +(probe (fn [p] (count (window {from: (mk p) part: [instrument] order: [rid] frame: 2 funcs: {prev: (first v) n: (count v)}}))) 5 60) -- true +;; no part/order: the whole table is one partition +(probe (fn [p] (count (window {from: (mk p) funcs: {n: (count v)}}))) 5 60) -- true +;; running frame +(probe (fn [p] (count (window {from: (mk p) part: [instrument] order: [rid] frame: 'running funcs: {s: (sum v)}}))) 5 60) -- true +;; a where: on the window input (from: is materialised before the window) +(probe (fn [p] (count (window {from: (select {from: (mk p) where: (> rid 3)}) part: [instrument] order: [rid] funcs: {n: (count v)}}))) 5 60) -- true +;; the same guard sat under sort and limit: a sorted or limited select over +;; a table built for the call leaked it the same way +(probe (fn [p] (count (select {from: (mk p) asc: rid}))) 5 60) -- true +(probe (fn [p] (count (select {from: (mk p) desc: v take: 5}))) 5 60) -- true +(probe (fn [p] (count (select {from: (mk p) where: (> rid 3) take: 5}))) 5 60) -- true +(probe (fn [p] (count (select {from: (mk p) where: (> rid 3) asc: v}))) 5 60) -- true +(probe (fn [p] (count (select {from: (mk p) by: instrument s: (sum v) asc: s}))) 5 60) -- true +(probe (fn [p] (count (select {from: (mk p) by: instrument s: (sum v) desc: s take: 3}))) 5 60) -- true +;; the same table reused across calls must not be released twice: the +;; result is still right after many calls +(set T (mk 1)) +(map (fn [p] (window {from: T part: [instrument] order: [rid] funcs: {n: (count v)}})) (til 50)) +(count T) -- 786 +(count (window {from: T part: [instrument] order: [rid] funcs: {n: (count v)}})) -- 786 +(== (sum (at (window {from: T part: [instrument] order: [rid] frame: 'running funcs: {s: (sum v)}}) 's)) (sum (at (window {from: T part: [instrument] order: [rid] frame: 'running funcs: {s: (sum v)}}) 's))) -- true From a2fe4960d46a529d1717df161a2a9d6217eeae1b Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Mon, 21 Sep 2026 11:16:20 +0300 Subject: [PATCH 2/3] test(lang): pin the refcount of a table across window, sort and limit queries Twenty calls of each shape over one global table; the table's refcount must be exactly what it was before, and the table still whole after. Fails on the unfixed executor at the first window shape. Co-Authored-By: Claude Fable 5.1 --- test/test_lang.c | 53 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/test/test_lang.c b/test/test_lang.c index 1aaa33c64..afbdf69d0 100644 --- a/test/test_lang.c +++ b/test/test_lang.c @@ -8703,6 +8703,58 @@ static test_result_t test_builtin_group_guid_rfl(void) { * the expression over the vocabulary bytes a chunk at a time. With the * chunk shrunk to 100 values, a 200-value vocabulary crosses chunk * boundaries; the groups and aggregates must equal the in-memory answer. */ +/* ---- Test: a query releases the table it was given ------------------- + * Every node the executor evaluates returns an owned reference, the + * constant table node a query is rooted on included. The window, sort + * and limit cases used to skip that release when the input equalled the + * graph's table, so each such query left one reference behind: the + * table's refcount climbed by one per call and, for a table built for + * the call, the whole table stayed allocated. Pin the refcount: after + * twenty calls of each shape it must be exactly what it was before. */ +static test_result_t test_select_releases_input_table(void) { + ray_t* setup = ray_eval_str( + "(do (set __rl_i (til 786)) " + " (set __rl_syms (as 'SYMBOL (map (fn [i] (format \"inst%\" i)) (til 700)))) " + " (set __rl_T (table [instrument rid v] (list (take __rl_syms 786) __rl_i (% (* 7 __rl_i) 101)))) 0)"); + TEST_ASSERT_NOT_NULL(setup); TEST_ASSERT_FALSE(RAY_IS_ERR(setup)); ray_release(setup); + ray_t* T = ray_eval_str("__rl_T"); /* one more ref: ours */ + TEST_ASSERT_NOT_NULL(T); TEST_ASSERT_FALSE(RAY_IS_ERR(T)); + TEST_ASSERT_EQ_I(T->type, RAY_TABLE); + uint32_t rc0 = ray_atomic_load(&T->rc); + static const char* const shapes[] = { + "(window {from: __rl_T part: [instrument] order: [rid] frame: 2 funcs: {n: (count v)}})", + "(window {from: __rl_T funcs: {n: (count v)}})", + "(window {from: __rl_T part: [instrument] order: [rid] frame: 'running funcs: {s: (sum v)}})", + "(select {from: __rl_T asc: rid})", + "(select {from: __rl_T desc: v take: 5})", + "(select {from: __rl_T where: (> rid 3) take: 5})", + "(select {from: __rl_T where: (> rid 3) asc: v})", + "(select {from: __rl_T by: instrument s: (sum v) asc: s})", + "(select {from: __rl_T by: instrument s: (sum v) desc: s take: 3})", + "(select {from: __rl_T take: 5})", + "(select {from: __rl_T where: (> rid 3)})", + }; + for (size_t k = 0; k < sizeof(shapes) / sizeof(shapes[0]); k++) { + for (int i = 0; i < 20; i++) { + ray_t* r = ray_eval_str(shapes[k]); + TEST_ASSERT_NOT_NULL(r); + TEST_ASSERT_FALSE(RAY_IS_ERR(r)); + TEST_ASSERT_EQ_I(r->type, RAY_TABLE); + ray_release(r); + } + if (ray_atomic_load(&T->rc) != rc0) + FAIL(shapes[k]); + } + /* the table is still whole after all of that */ + ray_t* n = ray_eval_str("(count __rl_T)"); + TEST_ASSERT_NOT_NULL(n); TEST_ASSERT_FALSE(RAY_IS_ERR(n)); + TEST_ASSERT_EQ_I(n->i64, 786); + ray_release(n); + ray_release(T); + ray_release(ray_eval_str("(set __rl_T 0) (set __rl_i 0) (set __rl_syms 0)")); + PASS(); +} + static test_result_t test_select_derived_key_file_chunks(void) { ray_t* r = ray_eval_str( "(do (set __dk_i (til 20000)) " @@ -9600,6 +9652,7 @@ const test_entry_t lang_entries[] = { { "lang/builtin/group_guid_rfl", test_builtin_group_guid_rfl, lang_setup, lang_teardown }, { "lang/builtin/group_empty_list", test_builtin_group_empty_and_list, lang_setup, lang_teardown }, { "lang/select/derived_key_file_chunks", test_select_derived_key_file_chunks, lang_setup, lang_teardown }, + { "lang/select/releases_input_table", test_select_releases_input_table, lang_setup, lang_teardown }, { "lang/temporal/extract_builtins_fn", test_temporal_extract_builtins_fn, lang_setup, lang_teardown }, { "lang/temporal/extract_time_atom", test_temporal_extract_time_atom, lang_setup, lang_teardown }, { "lang/temporal/extract_time_vector", test_temporal_extract_time_vector, lang_setup, lang_teardown }, From 2ccebde24b0c9d3614b7b649b2b631b34689d5d0 Mon Sep 17 00:00:00 2001 From: Serhii Savchuk Date: Mon, 21 Sep 2026 11:26:16 +0300 Subject: [PATCH 3/3] fix(exec): the reduction case owns its input too The same borrowed-query-table guard sat under the reductions; no child evaluates to the query table there today, so nothing leaked, but the premise is the one the sort, window and limit cases just dropped. Co-Authored-By: Claude Fable 5.1 --- src/ops/exec.c | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/ops/exec.c b/src/ops/exec.c index d762388dc..de07dbbe4 100644 --- a/src/ops/exec.c +++ b/src/ops/exec.c @@ -2025,7 +2025,9 @@ static ray_t* exec_node_inner(ray_graph_t* g, ray_op_t* op) { if (!input || RAY_IS_ERR(input)) return input; /* Compact lazy selection before reducing — filters may have * set g->selection without materializing a compacted table. */ - bool own_input = (input != g->table); + /* Owned, like every exec_node result (see OP_SORT); no child + * evaluates to a borrowed query table. */ + bool own_input = true; if (g->selection && input->type == RAY_TABLE) { ray_t* compacted = sel_compact(g, input, g->selection, NULL, 0); if (own_input) ray_release(input);