bill/AQuery
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

finialize demo

Browse Source
This commit is contained in:
bill
2023-02-27 10:51:12 +08:00
parent 05cca378e0
commit c944b5dfcf
8 changed files with 61 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
demo/demo.aquery
+17
View File
@@ -0,0 +1,17 @@
#!aquery
f demo/prep.a
exec
procedure demoi run
procedure democq run
procedure democa run
create trigger t action demoi interval 15000
exec
create trigger c on source action democa when democq
exec
# f demo/test.a
# exec
prompt.py
+3 -3
View File
@@ -501,11 +501,11 @@ def prompt(running = lambda:True, next = lambda:input('> '), state : Optional[Pr
else:
state.cfg.has_dll = 0
state.currstats.compile_time = state.currstats.stop()
cxt.post_exec_triggers()
if build_this:
state.set_ready()
while state.get_ready():
state.wait_engine()
cxt.post_exec_triggers()
# while state.get_ready():
# state.wait_engine()
state.currstats.need_print = True
continue
reconstruct/ast.py
+5 -6
View File
@@ -1109,7 +1109,7 @@ class create_trigger(ast_node):
if 'interval' in node: # executed periodically from server
self.type = self.Type.Interval
self.interval = node['interval']
send_to_server(f'TI{self.trigger_name}\0{self.action_name}\0{self.interval}')
self.context.queries.append(f'TI{self.trigger_name}\0{self.action_name}\0{self.interval}')
else: # executed from sql backend
self.type = self.Type.Callback
self.query_name = node['query']
@@ -1118,11 +1118,10 @@ class create_trigger(ast_node):
if self.procedure and self.table_name in self.context.tables_byname:
self.table = self.context.tables_byname[self.table_name]
self.table.triggers.add(self)
send_to_server(
self.context.queries.append(
f'TC{self.trigger_name}\0{self.table_name}\0'
f'{self.query_name}\0{self.action_name}'
)
)
else:
return
self.context.triggers[self.trigger_name] = self
@@ -1136,11 +1135,11 @@ class create_trigger(ast_node):
def execute(self):
from engine.utils import send_to_server
send_to_server(f'TA{self.query_name}\0{self.action_name}')
self.context.queries.append(f'TA{self.query_name}\0{self.action_name}')
def remove(self):
from engine.utils import send_to_server
send_to_server(f'TR{self.trigger_name}')
self.context.queries.append(f'TR{self.trigger_name}')
class drop_trigger(ast_node):
server/libaquery.cpp
+11 -17
View File
@@ -621,21 +621,15 @@ vector_type<std::string_view>::vector_type(const uint32_t size, void* data) :
//std::cout<<size << container[1];
}
void activate_callback_based_trigger(Context* context, const char* cmd)
{
const char* query_name = cmd + 2;
const char* action_name = query_name;
while (*action_name++);
if(auto q = get_procedure(cxt, query_name),
a = get_procedure(cxt, action_name);
q.name == nullptr || a.name == nullptr
)
printf("Warning: Invalid query or action name: %s %s",
query_name, action_name);
else {
auto query = AQ_DupObject(&q);
auto action = AQ_DupObject(&a);
cxt->ct_host->execute_trigger(query, action);
}
StoredProcedure
get_procedure(Context* cxt, const char* name) {
auto res = cxt->stored_proc.find(name);
if (res == cxt->stored_proc.end())
return { .cnt = 0,
.postproc_modules = 0,
.queries = nullptr,
.name = nullptr,
.__rt_loaded_modules = nullptr
};
return res->second;
}
server/libaquery.h
+1 -2
View File
@@ -291,6 +291,5 @@ inline _This_Type* AQ_DupObject(_This_Type* __val) {
#endif //__USE_STD_SEMAPHORE__
void print_monetdb_results(void* _srv, const char* sep, const char* end, uint32_t limit);
void activate_callback_based_trigger(Context* context, const char* cmd);
StoredProcedure get_procedure(Context* cxt, const char* name);
#endif
server/monetdb_conn.cpp
+1 -1
View File
@@ -382,7 +382,7 @@ int ExecuteStoredProcedureEx(const StoredProcedure *p, Context* cxt){
break;
case 'T' : {
if (p->queries[i][1] == 'N') {
cxt->ct_host->execute_trigger(p->queries[i] + 2, cxt);
cxt->ct_host->execute_trigger(p->queries[i] + 2);
}
}
break;
server/server.cpp
+20 -13
View File
@@ -97,24 +97,31 @@ have_hge() {
Context* _g_cxt;
StoredProcedure
get_procedure(Context* cxt, const char* name) {
auto res = cxt->stored_proc.find(name);
if (res == cxt->stored_proc.end())
return { .cnt = 0,
.postproc_modules = 0,
.queries = nullptr,
.name = nullptr,
.__rt_loaded_modules = nullptr
};
return res->second;
}
__AQEXPORT__(StoredProcedure)
get_procedure_ex(const char* name){
return get_procedure(_g_cxt, name);
}
void activate_callback_based_trigger(Context* context, const char* cmd)
{
const char* query_name = cmd + 2;
const char* action_name = query_name;
while (*action_name++);
if(auto q = get_procedure(context, query_name),
a = get_procedure(context, action_name);
q.name == nullptr || a.name == nullptr
)
printf("Warning: Invalid query or action name: %s %s",
query_name, action_name);
else {
auto query = AQ_DupObject(&q);
auto action = AQ_DupObject(&a);
context->ct_host->execute_trigger(query, action);
}
}
// This function contains heap allocations, free after use
template<class String_T>
char* to_lpstr(const String_T& str){
server/threading.cpp
+3 -3
View File
@@ -235,14 +235,14 @@ void CallbackBasedTriggerHost::execute_trigger(StoredProcedure* query, StoredPro
this->tp->enqueue_task(payload);
}
void execute_trigger(const char* trigger_name) {
void CallbackBasedTriggerHost::execute_trigger(const char* trigger_name) {
auto vt_triggers = static_cast<aq_map<std::string, CallbackBasedTrigger> *>(this->triggers);
auto ptr = vt_triggers->find(trigger_name);
if (ptr != vt_triggers->end()) {
auto& tr = ptr->second;
if (!tr.materialized) {
tr.query = new CallbackBasedTrigger(get_procedure(cxt, tr.query_name));
tr.action = new CallbackBasedTrigger(get_procedure(cxt, tr.action_name));
tr.query = new StoredProcedure(get_procedure(cxt, tr.query_name));
tr.action = new StoredProcedure(get_procedure(cxt, tr.action_name));
tr.materialized = true;
}
this->execute_trigger(AQ_DupObject(tr.query), AQ_DupObject(tr.action));