diff --git a/tests/meson.build b/tests/meson.build index 00481cc6b..5688442fb 100644 --- a/tests/meson.build +++ b/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) diff --git a/tests/test_join_spill.c b/tests/test_join_spill.c index a3b8163b5..052a052b2 100644 --- a/tests/test_join_spill.c +++ b/tests/test_join_spill.c @@ -11,6 +11,16 @@ #include #include +#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); diff --git a/tests/test_session_cold_relation.c b/tests/test_session_cold_relation.c index c152e22fb..5bba92a7d 100644 --- a/tests/test_session_cold_relation.c +++ b/tests/test_session_cold_relation.c @@ -13,6 +13,16 @@ #include #include +#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); diff --git a/tests/test_wirelog_advanced_spill.c b/tests/test_wirelog_advanced_spill.c index 274023e09..aec5a9bf3 100644 --- a/tests/test_wirelog_advanced_spill.c +++ b/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 diff --git a/wirelog/wirelog-advanced.h b/wirelog/wirelog-advanced.h index 3f5a2a611..40d818ae5 100644 --- a/wirelog/wirelog-advanced.h +++ b/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. *