Skip to content

fix(exec): release the input table of window, sort and limit queries #604

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
singaraiona merged 3 commits into from
Sep 21, 2026
Merged

fix(exec): release the input table of window, sort and limit queries #604

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
63d91b1
fix(exec): release the input table of window, sort and limit uncondit…
ser-vasilich Sep 21, 2026
a2fe496
test(lang): pin the refcount of a table across window, sort and limit…
ser-vasilich Sep 21, 2026
2ccebde
fix(exec): the reduction case owns its input too
ser-vasilich Sep 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 19 additions & 12 deletions src/ops/exec.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -2610,20 +2612,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;
}

Expand Down Expand Up @@ -2835,20 +2843,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;
}

Expand All @@ -2865,14 +2874,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;
}
Expand Down Expand Up @@ -2935,7 +2944,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;
Expand All @@ -2949,15 +2958,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 {
Expand Down
42 changes: 42 additions & 0 deletions test/rfl/mem/query_input_release.rfl
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
53 changes: 53 additions & 0 deletions test/test_lang.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)) "
Expand Down Expand Up @@ -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 },
Expand Down
Loading