WIP: group by optimizations

server/aggregations.h

@@ -186,6 +186,21 @@ decayed_t<VT, types::GetLongType<T>> sumw(uint32_t w, const VT<T>& arr) {
 	return ret;
 }
 
+template<class T, template<typename ...> class VT>
+void avgw(uint32_t w, const VT<T>& arr, 
+	decayed_t<VT, types::GetFPType<types::GetLongType<T>>>& ret) {
+	typedef types::GetFPType<types::GetLongType<T>> FPType;
+	const uint32_t& len = arr.size;
+	uint32_t i = 0;
+	types::GetLongType<T> s{};
+	w = w > len ? len : w;
+	if (len) s = ret[i++] = arr[0];
+	for (; i < w; ++i)
+		ret[i] = (s += arr[i]) / (FPType)(i + 1);
+	for (; i < len; ++i)
+		ret[i] = ret[i - 1] + (arr[i] - arr[i - w]) / (FPType)w;
+}
+
 template<class T, template<typename ...> class VT>
 decayed_t<VT, types::GetFPType<types::GetLongType<T>>> avgw(uint32_t w, const VT<T>& arr) {
 	typedef types::GetFPType<types::GetLongType<T>> FPType;

server/hasher.h

@@ -132,3 +132,7 @@ namespace ankerl::unordered_dense{
 	struct hash<std::tuple<Types...>> : public hasher<Types...>{ };
 }
 
+struct aq_hashtable_value_t{
+	uint32_t id;
+	uint32_t cnt;
+};

server/monetdb_conn.cpp

@@ -6,6 +6,8 @@
 #include "monetdb_conn.h"
 #include "monetdbe.h"
 #include "table.h"
+#include <thread>
+
 #undef ERROR
 #undef static_assert
 
@@ -86,7 +88,10 @@ void Server::connect(Context *cxt){
     }
 
     server = (monetdbe_database*)malloc(sizeof(monetdbe_database));
-    auto ret = monetdbe_open(server, nullptr, nullptr);
+    monetdbe_options ops;
+    AQ_ZeroMemory(ops);
+    ops.nr_threads = std::thread::hardware_concurrency();
+    auto ret = monetdbe_open(server, nullptr, &ops);
     if (ret == 0){
         status = true;
         this->server = server;

server/server.cpp

@@ -191,6 +191,21 @@ constexpr prt_fn_t monetdbe_prtfns[] = {
 	aq_to_chars<std::nullptr_t>
 };
 
+#ifndef __AQ_USE_THREADEDGC__
+void aq_init_gc(void *handle, Context* cxt)
+{
+    typedef void (*aq_gc_init_t) (Context*);
+    if (handle && cxt){
+        auto sym = dlsym(handle, "__AQ_Init_GC__");
+        if(sym){
+            ((aq_gc_init_t)sym)(cxt);
+        }
+    }
+}
+#else //__AQ_USE_THREADEDGC__
+#define aq_init_gc(h, c) 
+#endif //__AQ_USE_THREADEDGC__
+
 #include "monetdbe.h"
 #undef max
 #undef min
@@ -363,12 +378,7 @@ start:
                             recorded_queries.emplace_back(copy_lpstr("N"));
                         }
                         handle = dlopen(proc_name, RTLD_NOW);
-#ifndef __AQ_USE_THREADEDGC__
-                        {
-                            typedef void (*aq_gc_init_t) (Context*);
-                            ((aq_gc_init_t)dlsym(handle, "__AQ_Init_GC__"))(cxt);
-                        }
-#endif
+                        aq_init_gc(handle, cxt);
                         if (procedure_recording) {
                             recorded_libraries.emplace_back(handle);
                         }
@@ -474,11 +484,13 @@ start:
                                         p.__rt_loaded_modules = static_cast<void**>(
                                             malloc(sizeof(void*) * p.postproc_modules));
                                         for(uint32_t j = 0; j < p.postproc_modules; ++j){
-                                            auto pj = dlopen(p.name, RTLD_NOW);
+                                            auto pj = dlopen((procedure_root + p.name + std::to_string(j) + ".so").c_str(), RTLD_NOW);
                                             if (pj == nullptr){
                                                 printf("Error: failed to load module %s\n", p.name);
                                                 return true;
                                             }
+                                            aq_init_gc(pj, cxt);
+
                                             p.__rt_loaded_modules[j] = pj;
                                         }
                                     }
@@ -528,6 +540,7 @@ start:
                                         puts(p.queries[j-1]);
                                     }
                                     fclose(fp);
+                                    p.__rt_loaded_modules = 0;
                                     return load_modules(p);
                                 };
                                 switch(n_recvd[i][1]){

server/table.h

@@ -289,6 +289,7 @@ public:
 		uint32_t len = end - start;
 		return ColView<_Ty>(orig, idxs.subvec(start, end));
 	}
+
 	ColRef<_Ty> subvec_deep(uint32_t start, uint32_t end) const {
 		uint32_t len = end - start;
 		ColRef<_Ty> subvec(len);

server/unordered_dense.h

@@ -1059,6 +1059,18 @@ public:
         return do_insert_or_assign(std::move(key), std::forward<M>(mapped)).first;
     }
 
+    template <class K, class M>
+    auto hashtable_push(K&& key, M& mapped) {
+        ++ mapped.id;
+        ++ mapped.cnt;
+        auto it_isinserted = try_emplace(std::forward<K>(key), std::forward<M>(mapped));
+        if (!it_isinserted.second) {
+            --mapped.cnt;
+            return it_isinserted.first->second.id;
+        }
+        return mapped.id;
+    }
+
     template <typename K,
               typename M,
               typename Q = T,