fix(columnar): reset stratum frontiers on full snapshot re-eval

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();

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,97 @@ 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)
+{
+    TEST("query snapshots rebuild IDB rows after input changes");
+
+    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"
+        "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) {
+        FAIL("open failed");
+        return;
+    }
+    if (wirelog_easy_insert(s, "A", 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;
+    }
+
+    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;
+    }
+    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;
+    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 != 1) {
+        FAIL("input change must not duplicate Q rows");
+        wirelog_easy_close(s);
+        return;
+    }
+    wirelog_easy_close(s);
+    PASS();
+}
+
 /* ======================================================================== */
 /* Main                                                                     */
 /* ======================================================================== */
@@ -1853,6 +1945,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);

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) {

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,

wirelog/columnar/session.c

@@ -1683,6 +1683,9 @@ 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->pending_full_input_eval = true;
 
     return 0;
 }
@@ -2140,6 +2143,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 +2297,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 +2306,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);
 
@@ -2416,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);
@@ -2427,13 +2437,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 +2648,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.