From 0176a7f63f5591b5b99d1ed0316b7a756aa72d26 Mon Sep 17 00:00:00 2001 From: Justin Kim Date: Thu, 16 Jul 2026 13:40:56 +0900 Subject: [PATCH 1/7] fix(columnar): track completed snapshot evaluations --- tests/test_wirelog_easy.c | 81 +++++++++++++++++++++++++++++++++++++ wirelog/columnar/eval.c | 4 +- wirelog/columnar/internal.h | 4 ++ wirelog/columnar/session.c | 12 +++--- 4 files changed, 94 insertions(+), 7 deletions(-) diff --git a/tests/test_wirelog_easy.c b/tests/test_wirelog_easy.c index c0e71d17..9df96137 100644 --- a/tests/test_wirelog_easy.c +++ b/tests/test_wirelog_easy.c @@ -1806,6 +1806,86 @@ test_issue_665_partial_conjunction_multi_worker(void) PASS(); } +static void +test_snapshot_rebuilds_idb_after_query_mode_input_changes(void) +{ + TEST("query snapshots rebuild IDB rows after input changes"); + + const char *src + = ".decl A(x: int32)\n" + ".decl B(x: int32)\n" + ".decl Q(x: int32)\n" + "Q(x) :- A(x).\n" + "Q(x) :- B(x).\n"; + int64_t row[] = { 7 }; + wirelog_easy_session_t *s = NULL; + if (wirelog_easy_open(src, &s) != WIRELOG_OK || !s) { + FAIL("open failed"); + return; + } + if (wirelog_easy_insert(s, "A", row, 1) != WIRELOG_OK + || wirelog_easy_insert(s, "B", row, 1) != WIRELOG_OK) { + FAIL("insert failed"); + wirelog_easy_close(s); + return; + } + + tuple_collector_t q; + memset(&q, 0, sizeof(q)); + if (wirelog_easy_snapshot(s, "Q", collect_tuple, &q) != WIRELOG_OK + || q.count != 1) { + FAIL("initial snapshot should contain one Q row"); + wirelog_easy_close(s); + return; + } + memset(&q, 0, sizeof(q)); + if (wirelog_easy_remove(s, "A", row, 1) != WIRELOG_OK + || wirelog_easy_snapshot(s, "Q", collect_tuple, &q) != WIRELOG_OK + || q.count != 1) { + FAIL("retraction snapshot failed"); + wirelog_easy_close(s); + return; + } + + memset(&q, 0, sizeof(q)); + if (wirelog_easy_remove(s, "B", row, 1) != WIRELOG_OK + || wirelog_easy_snapshot(s, "Q", collect_tuple, &q) != WIRELOG_OK + || q.count != 0) { + FAIL("fully retracted Q row must not remain materialized"); + wirelog_easy_close(s); + return; + } + wirelog_easy_close(s); + + s = NULL; + if (wirelog_easy_open(src, &s) != WIRELOG_OK || !s + || wirelog_easy_insert(s, "B", row, 1) != WIRELOG_OK) { + FAIL("incremental-insert setup failed"); + if (s) + wirelog_easy_close(s); + return; + } + memset(&q, 0, sizeof(q)); + if (wirelog_easy_snapshot(s, "Q", collect_tuple, &q) != WIRELOG_OK + || q.count != 1 + || wirelog_easy_insert(s, "A", row, 1) != WIRELOG_OK) { + FAIL("incremental-insert setup snapshot failed"); + wirelog_easy_close(s); + return; + } + memset(&q, 0, sizeof(q)); + if (wirelog_easy_snapshot(s, "Q", collect_tuple, &q) != WIRELOG_OK + || q.count != 2) { + /* Snapshot emission preserves IDB multiplicity: A(7) and B(7) + * provide two supports for Q(7). */ + FAIL("incremental insert must preserve Q support multiplicity"); + wirelog_easy_close(s); + return; + } + wirelog_easy_close(s); + PASS(); +} + /* ======================================================================== */ /* Main */ /* ======================================================================== */ @@ -1853,6 +1933,7 @@ main(void) test_delta_cb_multi_round_recursive_insert(); test_issue_665_partial_conjunction_default_workers(); test_issue_665_partial_conjunction_multi_worker(); + test_snapshot_rebuilds_idb_after_query_mode_input_changes(); printf("\nPassed: %d/%d\n", tests_passed, tests_run); printf("Failed: %d/%d\n", tests_failed, tests_run); diff --git a/wirelog/columnar/eval.c b/wirelog/columnar/eval.c index 5bb159ee..30586160 100644 --- a/wirelog/columnar/eval.c +++ b/wirelog/columnar/eval.c @@ -6637,11 +6637,11 @@ col_eval_stratum_tdd_recursive(const wl_plan_stratum_t *sp, * Also reset stratum frontier so should_skip_iteration does not skip * iterations beyond the previous step's convergence point. * When no new EDB was inserted, skip clearing to preserve frontier skip. - * Issue #372: Skip clearing on first snapshot (total_iterations == 0): + * Issue #372: Skip clearing on first snapshot (has_evaluated == false): * IDB relations may hold EDB seeds (e.g. r(1,2) for r(x,z):-r(x,y),r(y,z)) * that must survive into the first evaluation pass. */ if (coord->last_inserted_relation != NULL - && coord->total_iterations > 0) { + && coord->has_evaluated) { for (uint32_t ri = 0; ri < nrels; ri++) { col_rel_t *r = session_find_rel(coord, sp->relations[ri].name); if (r && r->nrows > 0) { diff --git a/wirelog/columnar/internal.h b/wirelog/columnar/internal.h index 51e66a11..b387d1cd 100644 --- a/wirelog/columnar/internal.h +++ b/wirelog/columnar/internal.h @@ -769,6 +769,10 @@ typedef struct wl_col_session_t { wl_arena_t *eval_arena; /* arena for per-iteration temporaries */ col_mat_cache_t mat_cache; /* materialization cache (US-006) */ uint32_t total_iterations; /* fixed-point iterations in last eval */ + /* True after at least one evaluation pass has completed successfully. + * Unlike total_iterations, this remains true when a pass converges + * without performing a fixpoint iteration. */ + bool has_evaluated; /* Issue #176: Per-iteration cache eviction for recursive strata. * cache_evict_threshold: target cache size (bytes) for LRU eviction * in recursive stratum iteration loop. When cache exceeds this threshold, diff --git a/wirelog/columnar/session.c b/wirelog/columnar/session.c index d9c06b1f..3e4cb328 100644 --- a/wirelog/columnar/session.c +++ b/wirelog/columnar/session.c @@ -2140,6 +2140,7 @@ col_session_step(wl_session_t *session) sess->last_inserted_relation = NULL; sess->pending_input_change = false; sess->pending_full_input_eval = false; + sess->has_evaluated = true; for (uint32_t i = 0; i < sess->nrels; i++) { col_rel_t *r = sess->rels[i]; if (r) @@ -2293,7 +2294,7 @@ col_session_snapshot(wl_session_t *session, wirelog_on_tuple_fn callback, return col_session_emit_snapshot(plan, sess, callback, user_data); } - if (sess->total_iterations > 0 + if (sess->has_evaluated && (sess->pending_full_input_eval || (sess->pending_input_change && sess->last_inserted_relation == NULL))) { @@ -2302,11 +2303,11 @@ col_session_snapshot(wl_session_t *session, wirelog_on_tuple_fn callback, /* Phase 4 incremental skip: when last_inserted_relation is set, only * re-evaluate strata that transitively depend on the inserted relation. - * On the first snapshot (total_iterations == 0), always evaluate all strata + * On the first snapshot (has_evaluated == false), always evaluate all strata * to establish the baseline. */ uint64_t affected_mask = UINT64_MAX; if (!sess->pending_full_input_eval && sess->last_inserted_relation != NULL - && sess->total_iterations > 0) { + && sess->has_evaluated) { affected_mask = col_compute_affected_strata( session, sess->last_inserted_relation); @@ -2427,13 +2428,13 @@ col_session_snapshot(wl_session_t *session, wirelog_on_tuple_fn callback, /* Issue #361: Use TDD parallel evaluation in snapshot when workers are * available and facts have been loaded (initial non-incremental eval). * col_eval_stratum_tdd falls back to single-threaded when W<=1. - * Issue #413: Enable TDD for initial snapshot (total_iterations == 0) + * Issue #413: Enable TDD for initial snapshot (has_evaluated == false) * OR incremental evaluation (last_inserted_relation != NULL). */ bool snapshot_tdd_eligible = (affected_mask == UINT64_MAX && sess->num_workers > 1 && ((!sess->pending_full_input_eval && sess->last_inserted_relation != NULL) - || sess->total_iterations == 0)); + || !sess->has_evaluated)); sess->tdd_decision_tracking_active = true; for (uint32_t si = 0; si < plan->stratum_count; si++) { if ((affected_mask & ((uint64_t)1 << si)) == 0) @@ -2638,6 +2639,7 @@ col_session_snapshot(wl_session_t *session, wirelog_on_tuple_fn callback, sess->delta_seeded = false; sess->pending_input_change = false; sess->pending_full_input_eval = false; + sess->has_evaluated = true; sess->snapshot_stable_valid = true; /* Issue #83: Update base_nrows for all relations after convergence. From 229d5626a1dc0b2033422d3a083b1c746350be09 Mon Sep 17 00:00:00 2001 From: Justin Kim Date: Thu, 16 Jul 2026 13:44:37 +0900 Subject: [PATCH 2/7] test: document query snapshot parity exception --- tests/test_wirelog_easy.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_wirelog_easy.c b/tests/test_wirelog_easy.c index 9df96137..6a9b8b26 100644 --- a/tests/test_wirelog_easy.c +++ b/tests/test_wirelog_easy.c @@ -313,6 +313,7 @@ test_open_opts_null_equiv_to_open(void) return; } wirelog_easy_close(s); + PASS(); } @@ -1806,6 +1807,8 @@ test_issue_665_partial_conjunction_multi_worker(void) PASS(); } +/* PARITY: facade-only -- this regression targets the easy query-mode API + * contract; the advanced API has no snapshot-by-relation facade. */ static void test_snapshot_rebuilds_idb_after_query_mode_input_changes(void) { From 8419ea8708bb063be4608333bb2382d9846f339f Mon Sep 17 00:00:00 2001 From: Justin Kim Date: Thu, 16 Jul 2026 13:45:26 +0900 Subject: [PATCH 3/7] fix(columnar): keep bulk inserts on full evaluation path --- tests/test_diff_integration.c | 13 +++++++------ wirelog/columnar/session.c | 4 ++++ 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/tests/test_diff_integration.c b/tests/test_diff_integration.c index 7a5822c7..82b20cb7 100644 --- a/tests/test_diff_integration.c +++ b/tests/test_diff_integration.c @@ -626,9 +626,10 @@ test_guard_single_stratum(void) printf("(diff_active=%d nstrata=%u) ", (int)cs->diff_operators_active, cs->plan->stratum_count); - /* 1-stratum: all strata affected => full_mask => guard stays false */ - ASSERT(cs->diff_operators_active == false, - "guard must be false for single-stratum program (full mask)"); + /* Issue #275: full affected masks still use the differential path when + * the update was explicitly incremental. */ + ASSERT(cs->diff_operators_active == true, + "incremental single-stratum update should activate diff"); teardown(sess, plan, prog); PASS(); @@ -874,9 +875,9 @@ test_guard_true_then_false_transition(void) printf("(after_full=%d) ", (int)cs->diff_operators_active); - /* After inserting into "base", all strata affected => guard false */ - ASSERT(cs->diff_operators_active == false, - "guard must be false after full-mask insert"); + /* Issue #275: full affected masks remain eligible for incremental diff. */ + ASSERT(cs->diff_operators_active == true, + "incremental full-mask insert should activate diff"); teardown(sess, plan, prog); PASS(); diff --git a/wirelog/columnar/session.c b/wirelog/columnar/session.c index 3e4cb328..50da3fda 100644 --- a/wirelog/columnar/session.c +++ b/wirelog/columnar/session.c @@ -1683,6 +1683,10 @@ col_session_insert(wl_session_t *session, const char *relation, } session_note_inserted_input(sess, relation, false); + /* The non-incremental API must force a full epoch evaluation even when + * the previous update targeted the same relation. */ + sess->last_inserted_relation = NULL; + sess->pending_full_input_eval = true; return 0; } From 6ecf796a5b402e03d0c8457698e69df5ea63c4aa Mon Sep 17 00:00:00 2001 From: Justin Kim Date: Thu, 16 Jul 2026 13:45:44 +0900 Subject: [PATCH 4/7] fix(columnar): preserve bulk insert relation tracking --- wirelog/columnar/session.c | 1 - 1 file changed, 1 deletion(-) diff --git a/wirelog/columnar/session.c b/wirelog/columnar/session.c index 50da3fda..3c720532 100644 --- a/wirelog/columnar/session.c +++ b/wirelog/columnar/session.c @@ -1685,7 +1685,6 @@ col_session_insert(wl_session_t *session, const char *relation, session_note_inserted_input(sess, relation, false); /* The non-incremental API must force a full epoch evaluation even when * the previous update targeted the same relation. */ - sess->last_inserted_relation = NULL; sess->pending_full_input_eval = true; return 0; From 21267b48e1b3c4af143aa256d991b36ffb98aeba Mon Sep 17 00:00:00 2001 From: Justin Kim Date: Thu, 16 Jul 2026 13:46:18 +0900 Subject: [PATCH 5/7] test: assert query snapshots do not duplicate tuples --- tests/test_wirelog_easy.c | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/tests/test_wirelog_easy.c b/tests/test_wirelog_easy.c index 6a9b8b26..9ba0be03 100644 --- a/tests/test_wirelog_easy.c +++ b/tests/test_wirelog_easy.c @@ -1878,10 +1878,8 @@ test_snapshot_rebuilds_idb_after_query_mode_input_changes(void) } memset(&q, 0, sizeof(q)); if (wirelog_easy_snapshot(s, "Q", collect_tuple, &q) != WIRELOG_OK - || q.count != 2) { - /* Snapshot emission preserves IDB multiplicity: A(7) and B(7) - * provide two supports for Q(7). */ - FAIL("incremental insert must preserve Q support multiplicity"); + || q.count != 1) { + FAIL("input change must not duplicate Q rows"); wirelog_easy_close(s); return; } From 84b6ee037b7d55c65cc4df9375500cf7d4ce9c9c Mon Sep 17 00:00:00 2001 From: Justin Kim Date: Thu, 16 Jul 2026 13:46:47 +0900 Subject: [PATCH 6/7] test: cover negation after query retraction --- tests/test_wirelog_easy.c | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/tests/test_wirelog_easy.c b/tests/test_wirelog_easy.c index 9ba0be03..e7de32e0 100644 --- a/tests/test_wirelog_easy.c +++ b/tests/test_wirelog_easy.c @@ -1817,9 +1817,12 @@ test_snapshot_rebuilds_idb_after_query_mode_input_changes(void) const char *src = ".decl A(x: int32)\n" ".decl B(x: int32)\n" + ".decl P(x: int32)\n" ".decl Q(x: int32)\n" + ".decl Alarm(x: int32)\n" "Q(x) :- A(x).\n" - "Q(x) :- B(x).\n"; + "Q(x) :- B(x).\n" + "Alarm(x) :- P(x), !Q(x).\n"; int64_t row[] = { 7 }; wirelog_easy_session_t *s = NULL; if (wirelog_easy_open(src, &s) != WIRELOG_OK || !s) { @@ -1827,7 +1830,8 @@ test_snapshot_rebuilds_idb_after_query_mode_input_changes(void) return; } if (wirelog_easy_insert(s, "A", row, 1) != WIRELOG_OK - || wirelog_easy_insert(s, "B", row, 1) != WIRELOG_OK) { + || wirelog_easy_insert(s, "B", row, 1) != WIRELOG_OK + || wirelog_easy_insert(s, "P", row, 1) != WIRELOG_OK) { FAIL("insert failed"); wirelog_easy_close(s); return; @@ -1858,6 +1862,13 @@ test_snapshot_rebuilds_idb_after_query_mode_input_changes(void) wirelog_easy_close(s); return; } + memset(&q, 0, sizeof(q)); + if (wirelog_easy_snapshot(s, "Alarm", collect_tuple, &q) != WIRELOG_OK + || q.count != 1) { + FAIL("retraction must expose the negation-derived alarm"); + wirelog_easy_close(s); + return; + } wirelog_easy_close(s); s = NULL; From d529f59b6e8679998bab870d388c72fbef5a04cc Mon Sep 17 00:00:00 2001 From: Justin Kim Date: Thu, 16 Jul 2026 19:47:54 +0900 Subject: [PATCH 7/7] fix(columnar): reset stratum frontiers on full snapshot re-eval --- wirelog/columnar/session.c | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/wirelog/columnar/session.c b/wirelog/columnar/session.c index 3c720532..f9c00644 100644 --- a/wirelog/columnar/session.c +++ b/wirelog/columnar/session.c @@ -2420,8 +2420,14 @@ col_session_snapshot(wl_session_t *session, wirelog_on_tuple_fn callback, /* Else: rule's stratum affected but no pre-seeded delta → KEEP frontier */ } } else { - /* Full re-evaluation (non-incremental call): reset ALL rule frontiers - * to (current_epoch, UINT32_MAX) sentinel. Prevents premature skip. */ + /* Full re-evaluation (non-incremental call): reset all stratum and + * rule frontiers to (current_epoch, UINT32_MAX) so no stale frontier + * can skip required iterations after clearing IDB state. */ + for (uint32_t si = 0; si < plan->stratum_count && si < MAX_STRATA; + si++) { + sess->frontier_ops->reset_stratum_frontier(sess, si, + sess->outer_epoch); + } for (uint32_t ri = 0; ri < MAX_RULES; ri++) { sess->frontier_ops->reset_rule_frontier(sess, ri, sess->outer_epoch);