test(spill): clear the inherited join batch knob (Fixes #2287)

tests/meson.build

@@ -5745,6 +5745,10 @@ test_wirelog_advanced_spill_exe = executable(
 )
 test('wirelog_advanced_spill', test_wirelog_advanced_spill_exe,
   suite: ['unit', 'tsan'], timeout: 120)
+# #2287: the spill fixtures must not inherit the bounded join knob.
+test('wirelog_advanced_spill_join_batch_env', test_wirelog_advanced_spill_exe,
+  env: ['WIRELOG_JOIN_BATCH_BYTES=224'],
+  suite: ['unit'], timeout: 120)
 
 
 # Issue #841: umbrella public API symbols are real linkable/runtime
@@ -6590,6 +6594,10 @@ test_join_spill_exe = executable(
   link_with: [testlib_prod],
 )
 test('join_spill', test_join_spill_exe, suite: ['unit', 'tsan'], timeout: 120)
+# #2287: the spill fixtures must not inherit the bounded join knob.
+test('join_spill_join_batch_env', test_join_spill_exe,
+  env: ['WIRELOG_JOIN_BATCH_BYTES=224'],
+  suite: ['unit'], timeout: 120)
 
 # Production resumable observer baseline and transferable reader transitions.
 test_evaluation_observer_build_exe = executable(
@@ -6632,3 +6640,7 @@ test_session_cold_relation_exe = executable(
   link_with: [testlib_prod],
 )
 test('session_cold_relation', test_session_cold_relation_exe, suite: ['unit', 'tsan'], timeout: 120)
+# #2287: the spill fixtures must not inherit the bounded join knob.
+test('session_cold_relation_join_batch_env', test_session_cold_relation_exe,
+  env: ['WIRELOG_JOIN_BATCH_BYTES=224'],
+  suite: ['unit'], timeout: 120)

tests/test_join_spill.c

@@ -11,6 +11,16 @@
 #include <stdlib.h>
 #include <string.h>
 
+#ifdef _WIN32
+static int
+wl_test_unsetenv_(const char *name)
+{
+    return _putenv_s(name, "");
+}
+
+#define unsetenv wl_test_unsetenv_
+#endif
+
 #define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", __FILE__, \
                                       __LINE__, #x); abort(); } } while (0)
 #define BUDGET (UINT64_C(256) * 1024 * 1024)
@@ -1122,6 +1132,13 @@ fixed_budget_reclamation(void)
 int
 main(void)
 {
+    /* Issue #2287: session creation reads WIRELOG_JOIN_BATCH_BYTES and
+     * _STRICT, and a nonzero batch size refuses spill JOIN selection and
+     * sealed cold input. The knob cases here set the session fields
+     * directly, so clear the inherited environment before the first
+     * session exists. */
+    CHECK(!unsetenv("WIRELOG_JOIN_BATCH_BYTES"));
+    CHECK(!unsetenv("WIRELOG_JOIN_BATCH_STRICT"));
     for (unsigned enabled = 0; enabled < 2; enabled++) {
         fixture_t f = fixture(enabled != 0, 65536);
         wl_col_session_t *s = COL_SESSION(f.session);

tests/test_session_cold_relation.c

@@ -13,6 +13,16 @@
 #include <stdlib.h>
 #include <string.h>
 
+#ifdef _WIN32
+static int
+wl_test_unsetenv_(const char *name)
+{
+    return _putenv_s(name, "");
+}
+
+#define unsetenv wl_test_unsetenv_
+#endif
+
 #define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", \
                                       __FILE__, __LINE__, #x); abort(); \
                       } } while (0)
@@ -1209,6 +1219,13 @@ static void filtered_shape(void)
 
 int main(void)
 {
+    /* Issue #2287: session creation reads WIRELOG_JOIN_BATCH_BYTES and
+     * _STRICT, and a nonzero batch size refuses spill JOIN selection and
+     * sealed cold input. The knob cases here set the session fields
+     * directly, so clear the inherited environment before the first
+     * session exists. */
+    CHECK(!unsetenv("WIRELOG_JOIN_BATCH_BYTES"));
+    CHECK(!unsetenv("WIRELOG_JOIN_BATCH_STRICT"));
     projection_shape(); filtered_predicates(); filtered_shape();
     for (unsigned filtered = 0; filtered < 2; filtered++) {
         projection_consumers(filtered != 0); projection_failures(filtered != 0);

tests/test_wirelog_advanced_spill.c

@@ -401,6 +401,68 @@ static void output_floor(void)
     wirelog_session_destroy(s); wirelog_program_free(p);
     CHECK(live_files() == 0);
 }
+static wirelog_session_t *knob_session(wirelog_program_t *p,
+    const char *knob)
+{
+    wirelog_session_t *s = NULL;
+    wirelog_session_options_v1_t o = configured();
+    environment("WIRELOG_JOIN_BATCH_BYTES", knob);
+    CHECK(wirelog_session_create_with_options(p, WIRELOG_BACKEND_DEFAULT, 1,
+        &o, &s) == WIRELOG_OK);
+    populate(s, false);
+    return s;
+}
+static void knob_snapshot(wirelog_session_t *s)
+{
+    events_t e = {0};
+    CHECK(wirelog_session_snapshot(s, observe, &e) == WIRELOG_OK);
+    CHECK(e.count == 4 && e.seen == 15 && e.second == 0);
+}
+/* Issue #2287: WIRELOG_JOIN_BATCH_BYTES is read when the session is created.
+ * A value that turns it on refuses the seal before capture and leaves the
+ * source mutable; the other values below leave sealing available. On
+ * Windows, environment() deletes the variable for "", so that case repeats
+ * the unset one there. */
+static void join_batch_knob(void)
+{
+    static const char *const off[] = {
+        "0", "", "abc", "5x", "0x10", "18446744073709551616"
+    };
+    static const char *const on[] = {"4096", " 5", "-1"};
+    wirelog_program_t *p = parse(query);
+    for (unsigned i = 0; i < sizeof(off)/sizeof(off[0]); i++) {
+        wirelog_session_t *s = knob_session(p, off[i]);
+        CHECK(wirelog_session_seal_join_input(s, "b") == WIRELOG_OK);
+#if defined(_WIN32) || defined(__linux__)
+        CHECK(live_files() > 0);
+#endif
+        knob_snapshot(s);
+        wirelog_session_destroy(s);
+        CHECK(live_files() == 0);
+    }
+    for (unsigned i = 0; i < sizeof(on)/sizeof(on[0]); i++) {
+        wirelog_session_t *s = knob_session(p, on[i]);
+        CHECK(wirelog_session_seal_join_input(s, "b") == WIRELOG_ERR_EXEC);
+        CHECK(live_files() == 0);
+        /* Clearing the environment does not change a created session. */
+        environment("WIRELOG_JOIN_BATCH_BYTES", NULL);
+        CHECK(wirelog_session_seal_join_input(s, "b") == WIRELOG_ERR_EXEC);
+        int64_t row[] = {200, 300};
+        CHECK(wirelog_session_insert(s, "b", row, 1, 2) == WIRELOG_OK);
+        CHECK(wirelog_session_remove(s, "b", row, 1, 2) == WIRELOG_OK);
+        knob_snapshot(s);
+        wirelog_session_destroy(s);
+        CHECK(live_files() == 0);
+    }
+    /* Setting the environment after creation does not refuse the seal. */
+    wirelog_session_t *s = knob_session(p, NULL);
+    environment("WIRELOG_JOIN_BATCH_BYTES", "4096");
+    CHECK(wirelog_session_seal_join_input(s, "b") == WIRELOG_OK);
+    environment("WIRELOG_JOIN_BATCH_BYTES", NULL);
+    knob_snapshot(s);
+    wirelog_session_destroy(s); wirelog_program_free(p);
+    CHECK(live_files() == 0);
+}
 
 typedef struct { unsigned keys, joined, seen; uint64_t key_bits[2];
 } mixed_events_t;
@@ -492,6 +554,11 @@ static void unsupported_filters(void)
 
 int main(void)
 {
+    /* Issue #2287: a session created with a nonzero WIRELOG_JOIN_BATCH_BYTES
+     * refuses wirelog_session_seal_join_input, so clear the inherited knob
+     * and its _STRICT companion before the first session exists. */
+    environment("WIRELOG_JOIN_BATCH_BYTES", NULL);
+    environment("WIRELOG_JOIN_BATCH_STRICT", NULL);
     char *budget = copy_environment("WIRELOG_MEMORY_BUDGET");
     environment("WIRELOG_MEMORY_BUDGET", "268435456");
 #ifdef _WIN32
@@ -505,6 +572,7 @@ int main(void)
 #endif
     descriptors(); lifecycle(); refusals(); typed_projection(); inline_facts();
     output_floor(); mixed_projection(); unsupported_filters();
+    join_batch_knob();
 #ifdef _WIN32
     CHECK(_rmdir(directory) == 0);
 #else

wirelog/wirelog-advanced.h

@@ -175,6 +175,15 @@ wirelog_session_create_with_snapshot(wirelog_program_t *program,
  * WIRELOG_MEMORY_BUDGET policy still governs managed memory. Inputs initially
  * fit in memory and capture needs additional scratch; final output remains
  * resident. Spill files are ephemeral and session-owned, not reopenable data.
+ *
+ * Environment: WIRELOG_JOIN_BATCH_BYTES is read when the session is created.
+ * It is on when its whole value parses as a nonzero base-10 integer the way
+ * strtoull() reads it, so leading whitespace and a sign are accepted ("-1"
+ * turns it on). Any other value, including an empty one, "0", "0x10", "5x",
+ * "abc" or one out of range, leaves it off. When it is on, that session
+ * selects no spill JOIN even with positive join caps, and
+ * wirelog_session_seal_join_input() returns WIRELOG_ERR_EXEC. Changing the
+ * environment after creation does not affect an existing session.
  */
 typedef struct {
     uint32_t struct_size;
@@ -217,13 +226,17 @@ wirelog_session_create_with_options(wirelog_program_t *program,
  * when no row passes. Scan-only sealing and general cold pipelines
  * are unsupported. No right filter, timestamps, references,
  * opaque/materialized rewrites, worker/control/delta modes or delta callback.
- * Optimized plans may therefore be ineligible. Only one input may be sealed.
+ * Optimized plans may therefore be ineligible. The session must also not have
+ * been created with WIRELOG_JOIN_BATCH_BYTES on (see
+ * wirelog_session_options_v1_t). Only one input may be sealed.
  *
  * Success reclaims the resident source and makes all fact/compound mutation
  * unsupported for the entire session. The binding is reused by supported
  * STEP/SNAPSHOT calls until destruction; sealing cannot be undone. A refusal
  * preserves the original source and allows retry only while the session is
- * still eligible; an evaluation attempt permanently closes the seal window.
+ * still eligible, and a refusal because WIRELOG_JOIN_BATCH_BYTES was on lasts
+ * for the session's lifetime; an evaluation attempt permanently closes the
+ * seal window.
  * Calls are synchronous and must be externally serialized with all session
  * operations. No callback or cancellation hook is invoked by this public API.
  *