Merge branch 'master' of https://git.billsun.dev/bill/AQuery

Makefile

@@ -2,7 +2,7 @@ OS_SUPPORT =
 MonetDB_LIB = 
 MonetDB_INC = 
 Defines = 
-CXXFLAGS = --std=c++1z
+CXXFLAGS = --std=c++2a
 ifeq ($(AQ_DEBUG), 1)
 	OPTFLAGS = -g3 -fsanitize=address -fsanitize=leak
 	LINKFLAGS = 

README.md

@@ -304,10 +304,11 @@ SELECT * FROM my_table WHERE c1 > 10
 - [x] Query Optimization
   - [x] Selection/Order by push-down
   - [x] Join Optimization (Only in Hybrid Engine)
-  - [ ] Threaded GC
+  - [x] Threaded GC
 - [ ] Extensibility 
   - [x] UDFs (Hybrid Engine only)
   - [x] SDK and User Module 
+  - [x] Stored Procedures
   - [ ] Triggers 
 
 # Known Issues:
@@ -343,3 +344,7 @@ SELECT * FROM my_table WHERE c1 > 10
 
 - [MonetDB](https://www.monetdb.org) <br>
   License (Mozilla Public License): https://github.com/MonetDB/MonetDB/blob/master/license.txt
+
+- [ankerl::unordered_dense](https://github.com/martinus/unordered_dense)<br>
+  Author: Martin Ankerl <br>
+  License (MIT): http://opensource.org/licenses/MIT <br>

prompt.py

@@ -613,16 +613,21 @@ def prompt(running = lambda:True, next = lambda:input('> '), state : Optional[Pr
             elif q.startswith('procedure'):
                 qs = re.split(r'[ \t\r\n]', q)
                 procedure_help = '''Usage: procedure <procedure_name> [record|stop|run|remove|save|load]'''
-                send_to_server = lambda str: state.send(1, ctypes.c_char_p(bytes(str, 'utf-8')))
+                def send_to_server(payload : str): 
+                    state.payload = (ctypes.c_char_p*1)(ctypes.c_char_p(bytes(payload, 'utf-8')))
+                    state.cfg.has_dll = 0
+                    state.send(1, state.payload)
+                    state.set_ready()
                 if len(qs) > 2:
                     if qs[2].lower() =='record':
-                        if state.current_procedure != qs[1]:             
+                        if state.current_procedure is not None and state.current_procedure != qs[1]:             
                             print(f'Cannot record 2 procedures at the same time. Stop recording {state.current_procedure} first.')
-                        elif not state.current_procedure:
+                        elif state.current_procedure is None:
                             state.current_procedure = qs[1]
-                            send_to_server(f'R\0{qs[1]}', 'utf-8')
+                            send_to_server(f'R\0{qs[1]}')
                     elif qs[2].lower() == 'stop':
                         send_to_server(f'RT\0{qs[1]}')
+                        state.current_procedure = None
                     else:
                         if state.current_procedure:
                             print(f'Procedure manipulation commands are disallowed during procedure recording.')
@@ -635,6 +640,10 @@ def prompt(running = lambda:True, next = lambda:input('> '), state : Optional[Pr
                             send_to_server(f'RS\0{qs[1]}')
                         elif qs[2].lower() == 'load':
                             send_to_server(f'RL\0{qs[1]}')
+                     
+                elif len(qs) > 1:
+                    if qs[1].lower() == 'display':
+                        send_to_server(f'Rd\0')
                 else:
                     print(procedure_help)
                 continue

reconstruct/ast.py

@@ -383,12 +383,13 @@ class projection(ast_node):
         if self.group_node is not None and self.group_node.use_sp_gb:
             gb_vartable : Dict[str, Union[str, int]] = deepcopy(self.pyname2cname)
             gb_cexprs : List[str] = []
-            
+            gb_colnames : List[str] = []
             for key, val in proj_map.items():
                 col_name = 'col_' + base62uuid(6)
                 self.context.emitc(f'decltype(auto) {col_name} = {self.out_table.contextname_cpp}->get_col<{key}>();')
                 gb_cexprs.append((col_name, val[2]))
-            self.group_node.finalize(gb_cexprs, gb_vartable)
+                gb_colnames.append(col_name)
+            self.group_node.finalize(gb_cexprs, gb_vartable, gb_colnames)
         else:
             for i, (key, val) in enumerate(proj_map.items()):
                 if type(val[1]) is int:
@@ -536,7 +537,7 @@ class groupby_c(ast_node):
     
     def produce(self, node : List[Tuple[expr, Set[ColRef]]]):
         self.context.headers.add('"./server/hasher.h"')
-        self.context.headers.add('unordered_map')
+        # self.context.headers.add('unordered_map')
         self.group = 'g' + base62uuid(7)
         self.group_type = 'record_type' + base62uuid(7)
         self.datasource = self.proj.datasource
@@ -565,8 +566,9 @@ class groupby_c(ast_node):
             [f'{c}[{scanner_itname}]' for c in g_contents_list]
         )
         self.context.emitc(f'typedef record<{",".join(g_contents_decltype)}> {self.group_type};')
-        self.context.emitc(f'unordered_map<{self.group_type}, vector_type<uint32_t>, '
+        self.context.emitc(f'ankerl::unordered_dense::map<{self.group_type}, vector_type<uint32_t>, '
             f'transTypes<{self.group_type}, hasher>> {self.group};')
+        self.context.emitc(f'{self.group}.reserve({first_col}.size);')
         self.n_grps = len(self.glist)
         self.scanner = scan(self, first_col + '.size', it_name=scanner_itname)
         self.scanner.add(f'{self.group}[forward_as_tuple({g_contents})].emplace_back({self.scanner.it_var});')
@@ -581,7 +583,10 @@ class groupby_c(ast_node):
     #     gscanner.add(f'{self.datasource.cxt_name}->order_by<{assumption.result()}>(&{val_var});')
     #     gscanner.finalize()
         
-    def finalize(self, cexprs : List[Tuple[str, expr]], var_table : Dict[str, Union[str, int]]):
+    def finalize(self, cexprs : List[Tuple[str, expr]], var_table : Dict[str, Union[str, int]], col_names : List[str]):
+        for c in col_names:
+            self.context.emitc(f'{c}.reserve({self.group}.size());')
+        
         gscanner = scan(self, self.group, loop_style = 'for_each')
         key_var = 'key_'+base62uuid(7)
         val_var = 'val_'+base62uuid(7)
@@ -713,10 +718,10 @@ class groupby(ast_node):
                 #     self.parent.var_table.
                 self.parent.col_ext.update(l[1])    
                 
-    def finalize(self, cexprs : List[Tuple[str, expr]], var_table : Dict[str, Union[str, int]]):
+    def finalize(self, cexprs : List[Tuple[str, expr]], var_table : Dict[str, Union[str, int]], col_names : List[str]):
         if self.use_sp_gb:
             self.dedicated_gb = groupby_c(self.parent, self.dedicated_glist)
-            self.dedicated_gb.finalize(cexprs, var_table)
+            self.dedicated_gb.finalize(cexprs, var_table, col_names)
 
 
 class join(ast_node):

server/aggregations.h

@@ -31,7 +31,7 @@ double avg(const VT<T>& v) {
 
 template<class T, template<typename ...> class VT>
 VT<double> sqrt(const VT<T>& v) {
-	VT<double> ret{ v.size };
+	VT<double> ret(v.size);
 	for (uint32_t i = 0; i < v.size; ++i) {
 		ret[i] = sqrt(v[i]);
 	}
@@ -52,7 +52,7 @@ VT<T> truncate(const VT<T>& v, const uint32_t precision) {
 		return v.subvec_memcpy();
 	auto multiplier = pow(10, precision);
 	auto max_truncate = std::numeric_limits<T>::max()/multiplier;
-	VT<T> ret{ v.size };
+	VT<T> ret(v.size);
 	for (uint32_t i = 0; i < v.size; ++i) { // round or trunc??
 		ret[i] = v[i] < max_truncate ? round(v[i] * multiplier)/multiplier : v[i];
 	}
@@ -102,7 +102,7 @@ decayed_t<VT, T> maxs(const VT<T>& arr) {
 template<class T, template<typename ...> class VT>
 decayed_t<VT, T> minw(uint32_t w, const VT<T>& arr) {
 	const uint32_t& len = arr.size;
-	decayed_t<VT, T> ret{ len };
+	decayed_t<VT, T> ret(len);
 	std::deque<std::pair<T, uint32_t>> cache;
 	for (int i = 0; i < len; ++i) {
 		if (!cache.empty() && cache.front().second == i - w) cache.pop_front();
@@ -194,7 +194,7 @@ decayed_t<VT, types::GetFPType<types::GetLongType<T>>> avgw(uint32_t w, const VT
 	uint32_t i = 0;
 	types::GetLongType<T> s{};
 	w = w > len ? len : w;
-	if (len)	s = ret[i++] = arr[0];
+	if (len) s = ret[i++] = arr[0];
 	for (; i < w; ++i)
 		ret[i] = (s += arr[i]) / (FPType)(i + 1);
 	for (; i < len; ++i)

server/gc.h

@@ -16,6 +16,8 @@ private:;
 	std::atomic<uint32_t> alive_cnt;
 	std::atomic<uint64_t> current_size;
 	volatile bool lock;
+	using gc_deallocator_t = void (*)(void*);
+
 	// maybe use volatile std::thread::id instead
 protected:
 	void acquire_lock();
@@ -46,6 +48,12 @@ public:
 		terminate_daemon();
 	}
 	static GC* gc_handle;
+	template <class T>
+	constexpr static inline gc_deallocator_t _delete(T*){
+		return [](void* v){
+			delete (T*)v;
+		};
+	} 
     constexpr static void(*_free) (void*) = free;
 };
 

server/hasher.h

@@ -3,7 +3,10 @@
 #include <type_traits>
 #include <tuple>
 #include <functional>
+#include <string_view>
 #include "types.h"
+// #include "robin_hood.h"
+#include "unordered_dense.h"
 // only works for 64 bit systems
 namespace hasher_consts{
 	constexpr size_t _FNV_offset_basis = 14695981039346656037ULL;
@@ -16,7 +19,7 @@ inline size_t append_bytes(const unsigned char* _First) noexcept {
 		_Val ^= static_cast<size_t>(*_First);
 		_Val *= hasher_consts::_FNV_prime;
 	}
-
+	
 	return _Val;
 }
 
@@ -67,37 +70,44 @@ struct hasher {
 #else
 		#define _current_type current_type
 #endif
-		return std::hash<_current_type>()(std::get<i>(record)) ^ hashi<i + 1>(record);
+		return ankerl::unordered_dense::hash<_current_type>()(std::get<i>(record)) ^ hashi<i + 1>(record);
 	}
 	size_t operator()(const std::tuple<Types...>& record) const {
 		return hashi(record);
 	}
 };
+template <class T>
+struct hasher<T>{
+	size_t operator()(const std::tuple<T>& record) const {
+		return ankerl::unordered_dense::hash<T>()(std::get<0>(record));
+	}
+};
 
-
+namespace ankerl::unordered_dense{
-namespace std{
-
 	template<>
 	struct hash<astring_view> {
 		size_t operator()(const astring_view& _Keyval) const noexcept {
-			return append_bytes(_Keyval.str);
+			
+			return ankerl::unordered_dense::hash<std::string_view>()(_Keyval.rstr);
+			//return append_bytes(_Keyval.str);
+			
 		}
 	};
 
 	template<>
 	struct hash<types::date_t> {
 		size_t operator() (const types::date_t& _Keyval) const noexcept {
-			return std::hash<unsigned int>()(*(unsigned int*)(&_Keyval));
+			return ankerl::unordered_dense::hash<unsigned int>()(*(unsigned int*)(&_Keyval));
 		}
 	};
 
 	template<>
 	struct hash<types::time_t> {
 		size_t operator() (const types::time_t& _Keyval) const noexcept {
-			return std::hash<unsigned int>()(_Keyval.ms) ^ 
+			return ankerl::unordered_dense::hash<unsigned int>()(_Keyval.ms) ^ 
-			std::hash<unsigned char>()(_Keyval.seconds) ^
+			ankerl::unordered_dense::hash<unsigned char>()(_Keyval.seconds) ^
-			std::hash<unsigned char>()(_Keyval.minutes) ^
+			ankerl::unordered_dense::hash<unsigned char>()(_Keyval.minutes) ^
-			std::hash<unsigned char>()(_Keyval.hours)
+			ankerl::unordered_dense::hash<unsigned char>()(_Keyval.hours)
 			;
 		}
 	};
@@ -105,8 +115,8 @@ namespace std{
 	template<>
 	struct hash<types::timestamp_t>{
 		size_t operator() (const types::timestamp_t& _Keyval) const noexcept {
-			return std::hash<types::date_t>()(_Keyval.date) ^ 
+			return ankerl::unordered_dense::hash<types::date_t>()(_Keyval.date) ^ 
-				std::hash<types::time_t>()(_Keyval.time);
+				ankerl::unordered_dense::hash<types::time_t>()(_Keyval.time);
 		}
 	};
 #ifdef __SIZEOF_INT128__
@@ -114,12 +124,11 @@ namespace std{
 	template<>
 	struct hash<int128_struct>{
 		size_t operator() (const int128_struct& _Keyval) const noexcept {
-			return std::hash<uint64_t>()(_Keyval.__struct.low) ^ std::hash<uint64_t>()(_Keyval.__struct.high);
+			return ankerl::unordered_dense::hash<uint64_t>()(_Keyval.__struct.low) ^ ankerl::unordered_dense::hash<uint64_t>()(_Keyval.__struct.high);
 		}
 	};
 #endif
 	template <class ...Types>
 	struct hash<std::tuple<Types...>> : public hasher<Types...>{ };
-
 }
 

server/libaquery.cpp

@@ -9,6 +9,7 @@
 #include "utils.h"
 #include "libaquery.h"
 #include <random>
+#include "gc.h"
 
 char* gbuf = nullptr;
 
@@ -276,10 +277,29 @@ inline const char* str(const bool& v) {
 	return v ? "true" : "false";
 }
 
+class A{
+	public:
+	std::chrono::high_resolution_clock::time_point tp;
+	A(){
+		tp = std::chrono::high_resolution_clock::now();
+		printf("A %llx created.\n", tp.time_since_epoch().count());
+	}
+	~A() {
+		printf("A %llx died after %lldns.\n", tp.time_since_epoch().count(),
+		 (std::chrono::high_resolution_clock::now() - tp).count());
+	}
+};
 
 Context::Context() {
     current.memory_map = new std::unordered_map<void*, deallocator_t>;
-    init_session();
+#ifndef __AQ_USE_THREADEDGC__
+	this->gc = new GC();
+#endif
+	GC::gc_handle->reg(new A(), 6553600, [](void* a){
+		puts("deleting");
+		delete ((A*)a);
+	});
+	init_session();
 }
 
 Context::~Context() {
@@ -328,7 +348,6 @@ void* Context::get_module_function(const char* fname){
 // 	std::cout << ')';
 // }
 
-#include "gc.h"
 #include <utility>
 #include <thread>
 #ifndef __AQ_USE_THREADEDGC__
@@ -383,7 +402,7 @@ void GC::daemon() {
 
 	while (alive) {
 		if (running) {
-			if (current_size - max_size > 0 || 
+			if (uint64_t(current_size) > max_size || 
 				forceclean_timer > forced_clean) 
 			{
 				gc();

server/libaquery.h

@@ -73,6 +73,7 @@ struct StoredProcedure{
 	const char* name;
 	void **__rt_loaded_modules;
 };
+
 struct Context{
     typedef int (*printf_type) (const char *format, ...);
 
@@ -90,7 +91,7 @@ struct Context{
 #ifdef THREADING
 	void* thread_pool;
 #endif	
-#ifdef __AQ_THREADED_GC__
+#ifndef __AQ_USE_THREADEDGC__
 	void* gc;
 #endif
 	printf_type print = &printf;
@@ -109,10 +110,9 @@ struct Context{
 	void init_session();
 	void end_session();
 	void* get_module_function(const char*);
-	std::unordered_map<const char*, void*> tables;
+	std::unordered_map<std::string, void*> tables;
-    std::unordered_map<const char*, uColRef *> cols;
+    std::unordered_map<std::string, uColRef *> cols;
-    std::unordered_map<const char*, void*> loaded_modules;
+    std::unordered_map<std::string, StoredProcedure> stored_proc;
-    std::unordered_map<const char*, StoredProcedure> stored_proc;
 };
 
 
@@ -163,4 +163,10 @@ template<> char* aq_to_chars<types::date_t>(void* , char*);
 template<> char* aq_to_chars<types::time_t>(void* , char*);
 template<> char* aq_to_chars<types::timestamp_t>(void* , char*);
 typedef int (*code_snippet)(void*);
+
+template <class _This_Struct>
+inline void AQ_ZeroMemory(_This_Struct& __val) {
+	memset(&__val, 0, sizeof(_This_Struct));
+}
+
 #endif

server/server.cpp

@@ -162,6 +162,20 @@ __AQEXPORT__(bool) have_hge(){
 
 using prt_fn_t = char* (*)(void*, char*);
 
+// This function contains heap allocations, free after use
+template<class String_T>
+char* to_lpstr(const String_T& str){
+    auto ret = static_cast<char*>(malloc(str.size() + 1));
+    memcpy(ret, str.c_str(), str.size());
+    ret[str.size()] = '\0';
+    return ret;
+}
+char* copy_lpstr(const char* str){
+    auto len = strlen(str);
+    auto ret = static_cast<char*>(malloc(len + 1));
+    memcpy(ret, str, len + 1);
+    return ret;
+}
 
 constexpr prt_fn_t monetdbe_prtfns[] = {
 	aq_to_chars<bool>, aq_to_chars<int8_t>, aq_to_chars<int16_t>, aq_to_chars<int32_t>, 
@@ -271,7 +285,18 @@ int dll_main(int argc, char** argv, Context* cxt){
     aq_timer timer;
     Config *cfg = reinterpret_cast<Config *>(argv[0]);
     std::unordered_map<std::string, void*> user_module_map;
+    std::string pwd = std::filesystem::current_path().c_str();
+    auto sep = std::filesystem::path::preferred_separator;
+    pwd += sep;
+    std::string procedure_root = pwd + "procedures" + sep;
     std::string procedure_name = "";
+    StoredProcedure current_procedure;
+    vector_type<char *> recorded_queries;
+    vector_type<void *> recorded_libraries;
+    bool procedure_recording = false, 
+         procedure_replaying = false;
+    uint32_t procedure_module_cursor = 0;
+
     if (cxt->module_function_maps == nullptr)
         cxt->module_function_maps = new std::unordered_map<std::string, void*>();
     auto module_fn_map = 
@@ -292,12 +317,12 @@ int dll_main(int argc, char** argv, Context* cxt){
         puts(*(const char**)(alt_server->getCol(0)));
         cxt->alt_server = alt_server;
     }
-    bool rec = false;
     while(cfg->running){
         ENGINE_ACQUIRE();
         if (cfg->new_query) {
             cfg->stats.postproc_time = 0;
             cfg->stats.monet_time = 0;
+start:
 
             void *handle = nullptr;
             void *user_module_handle = nullptr;
@@ -307,7 +332,28 @@ int dll_main(int argc, char** argv, Context* cxt){
                 Server* server = reinterpret_cast<Server*>(cxt->alt_server);
                 if(n_recv > 0){
                     if (cfg->backend_type == BACKEND_AQuery || cfg->has_dll) {
-                        handle = dlopen("./dll.so", RTLD_NOW);
+                        const char* proc_name = "./dll.so";
+                        std::string dll_path;
+                        if (procedure_recording) {
+                            dll_path = procedure_root + 
+                                procedure_name + std::to_string(recorded_libraries.size) + ".so";
+                            
+                            try{
+                                if (std::filesystem::exists(dll_path))
+                                    std::filesystem::remove(dll_path);
+                                std::filesystem::copy_file(proc_name, dll_path);
+                            } catch(std::filesystem::filesystem_error& e){
+                                puts(e.what());
+                                dll_path = proc_name;
+                            }
+                            proc_name = dll_path.c_str();
+                            //if(recorded_libraries.size)
+                            recorded_queries.emplace_back(copy_lpstr("N"));
+                        }
+                        handle = dlopen(proc_name, RTLD_NOW);
+                        if (procedure_recording) {
+                            recorded_libraries.emplace_back(handle);
+                        }
                     }
                     for (const auto& module : user_module_map){
                         initialize_module(module.first.c_str(), module.second, cxt);
@@ -315,18 +361,24 @@ int dll_main(int argc, char** argv, Context* cxt){
                     cxt->init_session();
                     for(int i = 0; i < n_recv; ++i)
                     {
-                        //printf("%s, %d\n", n_recvd[i], n_recvd[i][0] == 'Q');
+                        printf("%s, %d\n", n_recvd[i], n_recvd[i][0] == 'Q');
                         switch(n_recvd[i][0]){
                         case 'Q': // SQL query for monetdbe
                             {
+                                if(procedure_recording){
+                                    recorded_queries.emplace_back(copy_lpstr(n_recvd[i]));
+                                }
                                 timer.reset();
                                 server->exec(n_recvd[i] + 1);
                                 cfg->stats.monet_time += timer.elapsed();
-                                // printf("Exec Q%d: %s", i, n_recvd[i]);
+                                printf("Exec Q%d: %s", i, n_recvd[i]);
                             }
                             break;
                         case 'P': // Postprocessing procedure 
                             if(handle && !server->haserror()) {
+                                if (procedure_recording) {
+                                    recorded_queries.emplace_back(copy_lpstr(n_recvd[i]));
+                                }
                                 code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, n_recvd[i]+1));
                                 timer.reset();
                                 c(cxt);
@@ -360,6 +412,12 @@ int dll_main(int argc, char** argv, Context* cxt){
                         case 'O':
                             {
                                 if(!server->haserror()){
+                                    if (procedure_recording){
+                                        char* buf = (char*) malloc (sizeof(char) * 6);
+                                        memcpy(buf, n_recvd[i], 5);
+                                        buf[5] = '\0';
+                                        recorded_queries.emplace_back(buf);
+                                    }
                                     uint32_t limit;
                                     memcpy(&limit, n_recvd[i] + 1, sizeof(uint32_t));
                                     if (limit == 0)
@@ -380,36 +438,112 @@ int dll_main(int argc, char** argv, Context* cxt){
                                 user_module_map.erase(it);
                             }
                             break;
+                        case 'N':
+                            {
+                                if(procedure_module_cursor < current_procedure.postproc_modules)
+                                    handle = current_procedure.__rt_loaded_modules[procedure_module_cursor++];
+                                printf("Load %i = %p\n", procedure_module_cursor, handle);
+                            }
+                            break;
                         case 'R': //recorded procedure
                             {
-                                auto proc_name = n_recvd[i] + 1;
+                                auto proc_name = n_recvd[i] + 2;
                                 proc_name = *proc_name?proc_name : proc_name + 1;
-                                const auto& load_modules = [](StoredProcedure &p){
+                                puts(proc_name);
+                                const auto& load_modules = [&](StoredProcedure &p) {
                                     if (!p.__rt_loaded_modules){
                                         p.__rt_loaded_modules = static_cast<void**>(
                                             malloc(sizeof(void*) * p.postproc_modules));
                                         for(uint32_t j = 0; j < p.postproc_modules; ++j){
-                                            p.__rt_loaded_modules[j] = dlopen(p.name, RTLD_NOW);
+                                            auto pj = dlopen(p.name, RTLD_NOW);
+                                            if (pj == nullptr){
+                                                printf("Error: failed to load module %s\n", p.name);
+                                                return true;
+                                            }
+                                            p.__rt_loaded_modules[j] = pj;
                                         }
                                     }
+                                    return false;
+                                };
+                                const auto& save_proc_tofile = [&](const StoredProcedure& p)  {
+                                    auto config_name = procedure_root + procedure_name + ".aqp";
+                                    auto fp = fopen(config_name.c_str(), "wb");
+                                    if (fp == nullptr){
+                                        printf("Error: failed to open file %s\n", config_name.c_str());
+                                        return true;
+                                    }
+                                    fwrite(&p.cnt, sizeof(p.cnt), 1, fp);
+                                    fwrite(&p.postproc_modules, sizeof(p.postproc_modules), 1, fp);
+                                    for(uint32_t j = 0; j < p.cnt; ++j){
+                                        auto current_query = p.queries[j];
+                                        auto len_query = strlen(current_query);
+                                        fwrite(current_query, len_query + 1, 1, fp);
+                                    }
+                                    fclose(fp);
+                                    return false;
+                                };
+                                const auto& load_proc_fromfile = [&](StoredProcedure& p)  {
+                                    auto config_name = procedure_root + p.name + ".aqp";
+                                    auto fp = fopen(config_name.c_str(), "rb");
+                                    if(fp == nullptr){
+                                        puts("ERROR: Procedure not found on disk.");
+                                        return false;
+                                    }
+                                    fread(&p.cnt, sizeof(p.cnt), 1, fp);
+                                    fread(&p.postproc_modules, sizeof(p.postproc_modules), 1, fp);
+                                    auto offset_now = ftell(fp);
+                                    fseek(fp, 0, SEEK_END);
+                                    auto queries_size = ftell(fp) - offset_now;
+                                    fseek(fp, offset_now, SEEK_SET);
+
+                                    p.queries = static_cast<char**>(malloc(sizeof(char*) * p.cnt));
+                                    p.queries[0] = static_cast<char*>(malloc(sizeof(char) * queries_size));
+                                    fread(&p.queries[0], queries_size, 1, fp);
+
+                                    for(uint32_t j = 1; j < p.cnt; ++j){
+                                        p.queries[j] = p.queries[j-1];
+                                        while(*p.queries[j] != '\0')
+                                            ++p.queries[j];
+                                    }
+                                    fclose(fp);
+                                    return load_modules(p);
                                 };
                                 switch(n_recvd[i][1]){
                                     case '\0':
+                                        current_procedure.name = copy_lpstr(proc_name);
+                                        AQ_ZeroMemory(current_procedure);
+                                        procedure_recording = true;
                                         procedure_name = proc_name;
                                     break;
                                     case 'T':
+                                        current_procedure.queries = recorded_queries.container;
+                                        current_procedure.cnt = recorded_queries.size;
+                                        current_procedure.name = copy_lpstr(proc_name);
+                                        current_procedure.postproc_modules = recorded_libraries.size;
+                                        current_procedure.__rt_loaded_modules = recorded_libraries.container;
+                                        AQ_ZeroMemory(recorded_queries);
+                                        AQ_ZeroMemory(recorded_libraries);
+                                        procedure_recording = false;
+                                        save_proc_tofile(current_procedure);
+                                        cxt->stored_proc.insert_or_assign(procedure_name, current_procedure);
                                         procedure_name = "";
                                     break;
                                     case 'E': // execute procedure
                                     {
-                                        auto _proc = cxt->stored_proc.find(procedure_name.c_str());
+                                        procedure_module_cursor = 0;
-                                        if (_proc == cxt->stored_proc.end())
+                                        auto _proc = cxt->stored_proc.find(proc_name);
-                                            printf("Procedure %s not found.\n", procedure_name.c_str());
+                                        if (_proc == cxt->stored_proc.end()){
+                                            printf("Procedure %s not found. Trying load from disk.\n", proc_name);
+                                            if (load_proc_fromfile(current_procedure)){
+                                                cxt->stored_proc.insert_or_assign(proc_name, current_procedure);
+                                            }
+                                        }
                                         else{
-                                            StoredProcedure &p = _proc->second;
+                                            current_procedure = _proc->second;
-                                            n_recv = p.cnt;
+                                            n_recv = current_procedure.cnt;
-                                            n_recvd = p.queries;
+                                            n_recvd = current_procedure.queries;
-                                            load_modules(p);
+                                            load_modules(current_procedure);
+                                            goto start; // yes, I know, refactor later!!
                                         }
                                     }
                                     break;
@@ -419,12 +553,24 @@ int dll_main(int argc, char** argv, Context* cxt){
                                     break;
                                     case 'L': //load procedure
                                     break;
+                                    case 'd': // display all procedures
+                                    for(const auto& p : cxt->stored_proc){
+                                        printf("Procedure: %s, %d queries, %d modules:\n", p.first.c_str(), 
+                                            p.second.cnt, p.second.postproc_modules);
+                                        for(uint32_t j = 0; j < p.second.cnt; ++j){
+                                            printf("\tQuery %d: %s\n", j, p.second.queries[j]);
+                                        }
+                                        puts("");
+                                    }
+                                    break;
                                 }
                             }
                             break;
                         }
                     }
-                    if(handle) {
+                    if(handle && 
+                        !procedure_replaying && !procedure_recording) {
+                        printf("Destroy %p\n", handle);
                         dlclose(handle);
                         handle = nullptr;
                     }
@@ -432,13 +578,11 @@ int dll_main(int argc, char** argv, Context* cxt){
                     cxt->end_session();
                     n_recv = 0;
                 }
-                if(server->last_error == nullptr){
+                if (server->last_error != nullptr) {
-                    // TODO: Add feedback to prompt.
+                    printf("Monetdbe Error: %s\n", server->last_error);
-                }   
-                else{
                     server->last_error = nullptr;
                     //goto finalize;
-                } 
+                }   
             }
             
             // puts(cfg->has_dll ? "true" : "false");
@@ -487,7 +631,7 @@ extern "C" int __DLLEXPORT__ main(int argc, char** argv) {
 #endif
    // puts("running");
    Context* cxt = new Context();
-   cxt->aquery_root_path = std::filesystem::current_path().c_str();
+   cxt->aquery_root_path = to_lpstr(std::filesystem::current_path().string());
    // cxt->log("%d %s\n", argc, argv[1]);
 
 #ifdef THREADING

server/table.h

@@ -145,9 +145,19 @@ public:
 	ColRef<_Ty>& operator =(ColRef<_Ty>&& vt) {
 		vector_type<_Ty>::operator=(std::move(vt));
 		return *this;
+	
 	}
-	ColView<_Ty> operator [](const vector_type<uint32_t>& idxs) const {
+	// ColView<_Ty> operator [](vector_type<uint32_t>& idxs) const {
-		return ColView<_Ty>(*this, idxs);
+	// 	return ColView<_Ty>(*this, std::move(idxs));
+	// }
+	// ColView<_Ty> operator [](const vector_type<uint32_t>& idxs) const {
+	// 	return ColView<_Ty>(*this, idxs);
+	// }
+	vector_type<_Ty> operator[](vector_type<uint32_t>& idxs) const {
+		vector_type<_Ty> ret(idxs.size);
+		for (uint32_t i = 0; i < idxs.size; ++i)
+			ret.container[i] = this->container[idxs[i]];
+		return ret;
 	}
 	vector_type<_Ty> operator [](const std::vector<bool>& idxs) const {
 		vector_type<_Ty> ret (this->size);
@@ -226,7 +236,7 @@ class ColView : public vector_base<_Ty> {
 public:
 	typedef ColRef<_Ty> Decayed_t;
 	const uint32_t size;
-	const ColRef<_Ty> orig;
+	const ColRef<_Ty>& orig;
 	vector_type<uint32_t> idxs;
 	ColView(const ColRef<_Ty>& orig, vector_type<uint32_t>&& idxs) : orig(orig), size(idxs.size), idxs(std::move(idxs)) {}
 	ColView(const ColRef<_Ty>& orig, const vector_type<uint32_t>& idxs) : orig(orig), idxs(idxs), size(idxs.size) {}
@@ -274,6 +284,7 @@ public:
 			ret[i] = orig[idxs[i]];
 		return ret;
 	}
+
 	ColView<_Ty> subvec(uint32_t start, uint32_t end) const {
 		uint32_t len = end - start;
 		return ColView<_Ty>(orig, idxs.subvec(start, end));

server/vector_type.hpp

@@ -36,7 +36,7 @@ public:
 		this->size = vt.size;
 		this->capacity = vt.capacity;
 		if (capacity) {
-			// puts("copy");
+			//puts("copy");
 			this->container = (_Ty*)malloc(size * sizeof(_Ty));
 			memcpy(container, vt.container, sizeof(_Ty) * size);
 		}
@@ -153,18 +153,34 @@ public:
 		else
 			return distinct_copy();
 	}
-	inline void grow() {
+	// TODO: think of situations where this is a temp!! (copy on write!!!)
-		if (size >= capacity) { // geometric growth
+	template <bool _grow = true>
-			uint32_t new_capacity = size + 1 + (size >> 1);
+	inline void grow(uint32_t sz = 0) {
-			_Ty* n_container = (_Ty*)malloc(new_capacity * sizeof(_Ty));
+		if constexpr (_grow)
-			memcpy(n_container, container, sizeof(_Ty) * size);
+			sz = this->size;
+		if (sz >= capacity) { // geometric growth
+			uint32_t new_capacity;
+			if constexpr (_grow)
+				new_capacity = size + 1 + (size >> 1);
+			else	
+				new_capacity = sz;
+
+			_Ty* n_container = (_Ty*)realloc(container, new_capacity * sizeof(_Ty));
+			// memcpy(n_container, container, sizeof(_Ty) * size);
 			memset(n_container + size, 0, sizeof(_Ty) * (new_capacity - size));
-			if (capacity)
+			// if (capacity)
-				free(container);
+			// 	free(container);
 			container = n_container;
 			capacity = new_capacity;
 		}
 	}
+	inline void resize(const uint32_t sz){
+		size = sz;
+		grow<false>(sz);
+	}
+	inline void reserve(const uint32_t sz){
+		grow<false>(sz);
+	}
 
 	void emplace_back(const _Ty& _val) {
 		grow();