fix: fix concurrency issue on Reducer (#7214)

JoanFM · GuyAv46 · web-flow · commit 718477883ca3 · 2025-11-04T17:10:53.000Z
* fix: fix concurrency issue on Reducer

* test: fix test distagg

* fix: make it static array

* fix: fix proper static

* Update src/module-init/module-init.c

* Update src/aggregate/reducer.c

Co-authored-by: GuyAv46 &lt;47632673+GuyAv46@users.noreply.github.com&gt;

---------

Co-authored-by: GuyAv46 &lt;47632673+GuyAv46@users.noreply.github.com&gt;
diff --git a/src/aggregate/reducer.c b/src/aggregate/reducer.c
@@ -14,52 +14,36 @@ typedef struct {
   ReducerFactory fn;
 } FuncEntry;
 
-static FuncEntry *globalRegistry = NULL;
+// Static registry of all builtin reducers - no runtime registration needed
+static const FuncEntry globalRegistry[] = {
+  {"COUNT", RDCRCount_New},
+  {"SUM", RDCRSum_New},
+  {"TOLIST", RDCRToList_New},
+  {"MIN", RDCRMin_New},
+  {"MAX", RDCRMax_New},
+  {"AVG", RDCRAvg_New},
+  {"COUNT_DISTINCT", RDCRCountDistinct_New},
+  {"COUNT_DISTINCTISH", RDCRCountDistinctish_New},
+  {"QUANTILE", RDCRQuantile_New},
+  {"STDDEV", RDCRStdDev_New},
+  {"FIRST_VALUE", RDCRFirstValue_New},
+  {"RANDOM_SAMPLE", RDCRRandomSample_New},
+  {"HLL", RDCRHLL_New},
+  {"HLL_SUM", RDCRHLLSum_New}
+};
 
-void RDCR_RegisterFactory(const char *name, ReducerFactory factory) {
-  FuncEntry ent = {.name = name, .fn = factory};
-  FuncEntry *tail = array_ensure_tail(&globalRegistry, FuncEntry);
-  *tail = ent;
-}
-
-static int isBuiltinsRegistered = 0;
+#define REGISTRY_SIZE 14
+static_assert(sizeof(globalRegistry) == sizeof(FuncEntry) * REGISTRY_SIZE);
 
 ReducerFactory RDCR_GetFactory(const char *name) {
-  if (!isBuiltinsRegistered) {
-    isBuiltinsRegistered = 1;
-    RDCR_RegisterBuiltins();
-  }
-  size_t n = array_len(globalRegistry);
-  for (size_t ii = 0; ii < n; ++ii) {
+  for (size_t ii = 0; ii < REGISTRY_SIZE; ++ii) {
     if (!strcasecmp(globalRegistry[ii].name, name)) {
       return globalRegistry[ii].fn;
     }
   }
   return NULL;
 }
 
-#define RDCR_XBUILTIN(X)                           \
-  X(RDCRCount_New, "COUNT")                        \
-  X(RDCRSum_New, "SUM")                            \
-  X(RDCRToList_New, "TOLIST")                      \
-  X(RDCRMin_New, "MIN")                            \
-  X(RDCRMax_New, "MAX")                            \
-  X(RDCRAvg_New, "AVG")                            \
-  X(RDCRCountDistinct_New, "COUNT_DISTINCT")       \
-  X(RDCRCountDistinctish_New, "COUNT_DISTINCTISH") \
-  X(RDCRQuantile_New, "QUANTILE")                  \
-  X(RDCRStdDev_New, "STDDEV")                      \
-  X(RDCRFirstValue_New, "FIRST_VALUE")             \
-  X(RDCRRandomSample_New, "RANDOM_SAMPLE")         \
-  X(RDCRHLL_New, "HLL")                            \
-  X(RDCRHLLSum_New, "HLL_SUM")
-
-void RDCR_RegisterBuiltins(void) {
-#define X(fn, n) RDCR_RegisterFactory(n, fn);
-  RDCR_XBUILTIN(X);
-#undef X
-}
-
 int ReducerOpts_GetKey(const ReducerOptions *options, const RLookupKey **out) {
   ArgsCursor *ac = options->args;
   const char *s;
diff --git a/src/aggregate/reducer.h b/src/aggregate/reducer.h
@@ -172,8 +172,6 @@ Reducer *RDCRHLLSum_New(const ReducerOptions *);
 
 typedef Reducer *(*ReducerFactory)(const ReducerOptions *);
 ReducerFactory RDCR_GetFactory(const char *name);
-void RDCR_RegisterFactory(const char *name, ReducerFactory factory);
-void RDCR_RegisterBuiltins(void);
 
 #ifdef __cplusplus
 }

Original file line number	Diff line number	Diff line change
`@@ -172,8 +172,6 @@ Reducer RDCRHLLSum_New(const ReducerOptions );`
`172`	`172`
`173`	`173`	`typedef Reducer (ReducerFactory)(const ReducerOptions *);`
`174`	`174`	`ReducerFactory RDCR_GetFactory(const char *name);`
`175`		`-void RDCR_RegisterFactory(const char *name, ReducerFactory factory);`
`176`		`-void RDCR_RegisterBuiltins(void);`
`177`	`175`
`178`	`176`	`#ifdef __cplusplus`
`179`	`177`	`}`