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

Single Table query part 1

Browse Source
This commit is contained in:
bill
2022-04-14 01:53:11 +08:00
parent 4dd571b3d2
commit c46ce578d4
25 changed files with 866 additions and 582 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engine/ast.py
+29 -17
View File
@@ -1,10 +1,11 @@
from operator import index
from engine.utils import base62uuid
from copy import copy
# replace column info with this later.
class ColRef:
def __init__(self, cname, _ty, cobj, cnt, table:'TableInfo', name, id, compound = False):
self.cname = cname
self.cxt_name = None
self.cname = cname # column object location
self.cxt_name = None # column object in context
self.type = _ty
self.cobj = cobj
self.cnt = cnt
@@ -28,6 +29,7 @@ class ColRef:
base_name = f'{base_name}_{counter}'
self.cxt_name = base_name
cxt.columns_in_context[self] = base_name
# TODO: change this to cname;
cxt.emit(f'auto& {base_name} = *(ColRef<{self.type}> *)(&{self.table.cxt_name}->colrefs[{self.id}]);')
elif self.cxt_name is None:
self.cxt_name = cxt.columns_in_context[self]
@@ -44,7 +46,9 @@ class ColRef:
self.__arr__[key] = value
def __str__(self):
return self.cname
return self.reference()
def __repr__(self):
return self.reference()
class TableInfo:
@@ -87,6 +91,7 @@ class TableInfo:
self.cxt.emit(f'auto& {base_name} = *(TableInfo{type_tags} *)(cxt->tables[{self.table_name}]);')
def refer_all(self):
self.reference()
for c in self.columns:
c.reference()
def add_cols(self, cols, new = True):
@@ -95,12 +100,13 @@ class TableInfo:
def add_col(self, c, new = True, i = 0):
_ty = c['type']
if new:
cname =f'{self.table_name}->colrefs[{i}].scast<int>()'
cname =f'get<{i}>({self.table_name})'
_ty = _ty if type(c) is ColRef else list(_ty.keys())[0]
col_object = ColRef(cname, _ty, c, 1, self,c['name'], len(self.columns))
else:
col_object = c
cname = c.cname
c.table = self
self.cxt.ccols_byname[cname] = col_object
self.columns_byname[c['name']] = col_object
self.columns.append(col_object)
@@ -156,7 +162,7 @@ class TableInfo:
self.cxt.tables_byname[alias] = self
self.alias.add(alias)
def parse_tablenames(self, colExpr, materialize = True):
def parse_tablenames(self, colExpr, materialize = True, raw = False):
self.get_col = self.get_col if materialize else self.get_col_d
parsedColExpr = colExpr.split('.')
@@ -168,13 +174,12 @@ class TableInfo:
if datasource is None:
raise ValueError(f'Table name/alias not defined{parsedColExpr[0]}')
else:
ret = datasource.get_col(parsedColExpr[1])
if self.groupinfo is not None and ret:
ret = f"{ret.reference()}[{'start' if ret in self.groupinfo.referenced else 'range'}]"
else:
ret = ret.reference()
return ret
ret = datasource.parse_tablenames(parsedColExpr[1], raw)
from engine.expr import index_expr
string = ret.reference() + index_expr
if self.groupinfo is not None and ret and ret in self.groupinfo.raw_groups:
string = f'get<{self.groupinfo.raw_groups.index(ret)}>({{y}})'
return string, ret if raw else string
class View:
def __init__(self, context, table = None, tmp = True):
self.table: TableInfo = table
@@ -187,7 +192,11 @@ class View:
self.context.emit(f'{self.name}:()')
class Context:
function_head = 'extern \"C\" int dllmain(Context* cxt){ \n'
function_head = '''
extern "C" int __DLLEXPORT__ dllmain(Context* cxt) {
using namespace std;
using namespace types;
'''
def __init__(self):
self.tables:List[TableInfo] = []
self.tables_byname = dict()
@@ -208,7 +217,8 @@ class Context:
# and will be deactivated when the `from' is out of scope
self.datasource = None
self.ds_stack = []
self.scans = []
self.removing_scan = False
def add_table(self, table_name, cols):
tbl = TableInfo(table_name, cols, self)
self.tables.append(tbl)
@@ -257,6 +267,10 @@ class Context:
return ds
else:
return None
def remove_scan(self, scan, str_scan):
self.emit(str_scan)
self.scans.remove(scan)
def finalize(self):
if not self.finalized:
headers = ''
@@ -282,8 +296,6 @@ class ast_node:
self.context = parent.context if context is None else context
self.parent = parent
self.datasource = None
for h in self.header:
self.context.headers.add(h)
self.init(node)
self.produce(node)
self.spawn(node)
engine/ddl.py
+38 -17
View File
@@ -1,21 +1,35 @@
# code-gen for data decl languages
from engine.ast import ColRef, TableInfo, ast_node, include
from engine.ast import ColRef, TableInfo, ast_node, Context, include
from engine.utils import base62uuid
class create_table(ast_node):
name = 'create_table'
def __init__(self, parent: "ast_node", node, context: Context = None, cexpr = None):
self.cexpr = cexpr
super().__init__(parent, node, context)
def produce(self, node):
ct = node[self.name]
tbl = self.context.add_table(ct['name'], ct['columns'])
if type(node) is not TableInfo:
ct = node[self.name]
tbl = self.context.add_table(ct['name'], ct['columns'])
else:
tbl = node
col_type_str = ','.join([c.type for c in tbl.columns])
# create tables in c
self.emit(f"auto {tbl.table_name} = new TableInfo(\"{tbl.table_name}\", {tbl.n_cols});")
self.emit(f"auto {tbl.table_name} = new TableInfo<{col_type_str}>(\"{tbl.table_name}\", {tbl.n_cols});")
self.emit("cxt->tables.insert({\"" + tbl.table_name + f"\", {tbl.table_name}"+"});")
self.context.tables_in_context[tbl] = tbl.table_name
tbl.cxt_name = tbl.table_name
for i, c in enumerate(ct['columns']):
# TODO: more self awareness
self.emit(f"{tbl.table_name}->colrefs[{i}].ty = types::AINT;")
tbl.refer_all()
if self.cexpr is None:
for c in tbl.columns:
self.emit(f"{c.cxt_name}.init();")
else:
for i, c in enumerate(tbl.columns):
self.emit(f"{c.cxt_name}.init();")
self.emit(f"{c.cxt_name} = {self.cexpr[i]()};")
class insert(ast_node):
name = 'insert'
def produce(self, node):
@@ -42,18 +56,25 @@ class c(ast_node):
class load(ast_node):
name="load"
def produce(self, node):
self.context.headers.add('"csv.h"')
node = node[self.name]
table:TableInfo = self.context.tables_byname[node['table']]
n_keys = len(table.columns)
keys = ''
for _ in range(n_keys):
keys+='`tk'+base62uuid(6)
tablename = 'l'+base62uuid(7)
self.emit(f"{tablename}:({keys}!(+(`csv ? 1:\"{node['file']['literal']}\")))[{keys}]")
table.refer_all()
csv_reader_name = 'csv_reader_' + base62uuid(6)
col_types = [c.type for c in table.columns]
col_tmp_names = ['tmp_'+base62uuid(8) for _ in range(len(table.columns))]
# col_type_str = ",".join(col_types)
col_names = ','.join([f'"{c.name}"' for c in table.columns])
self.emit(f'io::CSVReader<{len(col_types)}> {csv_reader_name}("{node["file"]["literal"]}");')
self.emit(f'{csv_reader_name}.read_header(io::ignore_extra_column, {col_names});')
for t, n in zip(col_types, col_tmp_names):
self.emit(f'{t} {n};')
self.emit(f'while({csv_reader_name}.read_row({",".join(col_tmp_names)})) {{ \n')
for i, c in enumerate(table.columns):
self.emit(f'{c.cname}:{tablename}[{i}]')
self.emit(f'{c.cxt_name}.emplace_back({col_tmp_names[i]});')
self.emit('}')
class outfile(ast_node):
name="_outfile"
engine/expr.py
+39 -24
View File
@@ -1,5 +1,7 @@
from engine.ast import ast_node
from engine.ast import ast_node, ColRef
start_expr = 'f"'
index_expr = '{\'\' if x is None and y is None else f\'[{x}]\'}'
end_expr = '"'
class expr(ast_node):
name='expr'
@@ -8,7 +10,7 @@ class expr(ast_node):
'min': 'min',
'avg': 'avg',
'sum': 'sum',
'mod':'mod',
'count' : 'count',
'mins': ['mins', 'minsw'],
'maxs': ['maxs', 'maxsw'],
'avgs': ['avgs', 'avgsw'],
@@ -19,29 +21,32 @@ class expr(ast_node):
'sub':'-',
'add':'+',
'mul':'*',
'div':'%',
'div':'/',
'mod':'%',
'and':'&',
'or':'|',
'xor' : '^',
'gt':'>',
'lt':'<',
'le':'<=',
'gt':'>='
}
compound_ops = {
'ge' : [2, lambda x: f'~({x[0]}<{x[1]})'],
'le' : [2, lambda x: f'~({x[0]}>{x[1]})'],
'count' : [1, lambda x: f'#({x[0]})']
}
unary_ops = {
'neg' : '-',
'not' : '~'
'not' : '!'
}
coumpound_generating_ops = ['mod', 'mins', 'maxs', 'sums'] + \
list( binary_ops.keys()) + list(compound_ops.keys()) + list(unary_ops.keys() )
def __init__(self, parent, node, materialize_cols = True):
def __init__(self, parent, node, materialize_cols = True, abs_col = False):
self.materialize_cols = materialize_cols
self.raw_col = None
self.__abs = abs_col
ast_node.__init__(self, parent, node, None)
def init(self, _):
@@ -54,7 +59,8 @@ class expr(ast_node):
else:
self.datasource = self.context.datasource
self.udf_map = parent.context.udf_map
self.cexpr = ''
self._expr = ''
self.cexpr = None
self.func_maps = {**self.udf_map, **self.builtin_func_maps}
def produce(self, node):
@@ -62,30 +68,31 @@ class expr(ast_node):
for key, val in node.items():
if key in self.func_maps:
# if type(val) in [dict, str]:
self.context.headers.add('"./server/aggregations.h"')
if type(val) is list and len(val) > 1:
cfunc = self.func_maps[key]
cfunc = cfunc[len(val) - 1] if type(cfunc) is list else cfunc
self.cexpr += f"{cfunc}("
self._expr += f"{cfunc}("
for i, p in enumerate(val):
self.cexpr += expr(self, p).cexpr + (';'if i<len(val)-1 else '')
self._expr += expr(self, p)._expr + (','if i<len(val)-1 else '')
else:
funcname = self.func_maps[key]
funcname = funcname[0] if type(funcname) is list else funcname
self.cexpr += f"{funcname}("
self.cexpr += expr(self, val).cexpr
self.cexpr += ')'
self._expr += f"{funcname}("
self._expr += expr(self, val)._expr
self._expr += ')'
elif key in self.binary_ops:
l = expr(self, val[0]).cexpr
r = expr(self, val[1]).cexpr
self.cexpr += f'({l}{self.binary_ops[key]}{r})'
l = expr(self, val[0])._expr
r = expr(self, val[1])._expr
self._expr += f'({l}{self.binary_ops[key]}{r})'
elif key in self.compound_ops:
x = []
if type(val) is list:
for v in val:
x.append(expr(self, v).cexpr)
self.cexpr = self.compound_ops[key][1](x)
x.append(expr(self, v)._expr)
self._expr = self.compound_ops[key][1](x)
elif key in self.unary_ops:
self.cexpr += f'({expr(self, val).cexpr}{self.unary_ops[key]})'
self._expr += f'({expr(self, val)._expr}{self.unary_ops[key]})'
else:
print(f'Undefined expr: {key}{val}')
@@ -101,10 +108,18 @@ class expr(ast_node):
while type(p) is expr and not p.isvector:
p.isvector = True
p = p.parent
self.cexpr = self.datasource.parse_tablenames(node, self.materialize_cols)
self._expr, self.raw_col = self.datasource.parse_tablenames(node, self.materialize_cols, True)
self.raw_col = self.raw_col if type(self.raw_col) is ColRef else None
if self.__abs and self.raw_col:
self._expr = self.raw_col.reference() + index_expr
elif type(node) is bool:
self.cexpr = '1' if node else '0'
self._expr = '1' if node else '0'
else:
self.cexpr = f'{node}'
self._expr = f'{node}'
def toCExpr(_expr):
return lambda x = None, y = None : eval(start_expr + _expr + end_expr)
def consume(self, _):
self.cexpr = expr.toCExpr(self._expr)
def __str__(self):
return self.cexpr
engine/groupby.py
+35 -30
View File
@@ -1,56 +1,61 @@
from engine.ast import ast_node
from engine.ast import TableInfo, ast_node
from engine.scan import scan
from engine.utils import base62uuid
from engine.expr import expr
import engine.types
class groupby(ast_node):
name = '_groupby'
def init(self, _):
self.context.headers.add('"./server/hasher.h"')
self.context.headers.add('unordered_map')
self.group = 'g' + base62uuid(7)
self.group_type = 'record_type' + base62uuid(7)
self.datasource = self.parent.datasource
self.scanner = None
self.datasource.rec = []
self.raw_groups = []
def produce(self, node):
if type(node) is not list:
node = [node]
g_contents = '('
g_contents = ''
g_contents_list = []
first_col = ''
for i, g in enumerate(node):
v = g['value']
e = expr(self, v).cexpr
e = expr(self, v)
self.raw_groups.append(e.raw_col)
e = e._expr
# if v is compound expr, create tmp cols
if type(v) is not str:
tmpcol = 't' + base62uuid(7)
self.emit(f'{tmpcol}:{e}')
self.emit(f'auto {tmpcol} = {e};')
e = tmpcol
if i == 0:
first_col = e
g_contents += e + (';'if i < len(node)-1 else '')
self.emit(f'{self.group}:'+g_contents+')')
g_contents_list.append(e)
g_contents_decltype = [f'decltype({c})' for c in g_contents_list]
g_contents = expr.toCExpr(','.join(g_contents_list))
self.emit(f'typedef record<{expr.toCExpr(",".join(g_contents_decltype))(0)}> {self.group_type};')
self.emit(f'unordered_map<{self.group_type}, vector_type<uint32_t>, '
f'transTypes<{self.group_type}, hasher>> {self.group};')
self.n_grps = len(node)
if self.n_grps <= 1:
self.emit(f'{self.group}:={self.group}')
else:
self.emit(f'{self.group}:groupby[({self.group},(,!(#({first_col}))))]')
self.scanner = scan(self, None, expr.toCExpr(first_col)()+'.size')
self.scanner.add(f'{self.group}[forward_as_tuple({g_contents(self.scanner.it_ver)})].emplace_back({self.scanner.it_ver});')
def consume(self, _):
self.referenced = self.datasource.rec
self.datasource.rec = None
return super().consume(_)
self.scanner.finalize()
def finalize(self, ret, out):
self.groupby_function = 'fgrp'+base62uuid(4)
grp = self.group
if self.n_grps <= 1:
cfn = "{[range] start:*range;"+ ret + "}"
self.emit(f'{out}:(({cfn}\'{grp})[!{grp}])')
self.parent.inv = False
else:
cfn = "{[ids;grps;ll;dim;x] " + \
"start:grps[x][dim];" + \
"end:$[x=0;ll;grps[x-1][dim]];" + \
"range:(end-start)#((start-ll)#ids);" + \
"start:ids[start];" + \
ret + '}'
self.emit(f'{self.groupby_function}:{cfn}')
self.emit(f'{out}:+({self.groupby_function}' + \
f'[{grp}[1];{grp}[0];(#{grp}[0])+1;(#({grp}[0][0]))-1]\'!(#({grp}[0])))')
def finalize(self, cexprs, out:TableInfo):
gscanner = scan(self, self.group)
key_var = 'key_'+base62uuid(7)
val_var = 'val_'+base62uuid(7)
gscanner.add(f'auto &{key_var} = {gscanner.it_ver}.first;')
gscanner.add(f'auto &{val_var} = {gscanner.it_ver}.second;')
gscanner.add(';\n'.join([f'{out.columns[i].reference()}.emplace_back({ce(x=val_var, y=key_var)})' for i, ce in enumerate(cexprs)])+';')
gscanner.finalize()
engine/projection.py
+14 -22
View File
@@ -5,7 +5,7 @@ from engine.expr import expr
from engine.orderby import orderby
from engine.scan import filter
from engine.utils import base62uuid, enlist, base62alp
from engine.ddl import outfile
from engine.ddl import create_table, outfile
import copy
class projection(ast_node):
@@ -75,7 +75,7 @@ class projection(ast_node):
if self.group_node is not None:
# There is group by;
has_groupby = True
cexpr = f'('
cexprs = []
flatten = False
cols = []
self.out_table = TableInfo('out_'+base62uuid(4), [], self.context)
@@ -89,27 +89,23 @@ class projection(ast_node):
if type(proj) is dict:
if 'value' in proj:
e = proj['value']
if type(e) is str:
cname = e # TODO: deal w/ alias
cexpr += (f"{self.datasource.parse_tablenames(proj['value'])}")
elif type(e) is dict:
p_expr = expr(self, e)
cname = p_expr.cexpr
compound = True
cexpr += f"{cname}"
cname = ''.join([a if a in base62alp else '' for a in cname])
cexpr += ';'if i < len(self.projections)-1 else ''
sname = expr(self, e)._expr
fname = expr.toCExpr(sname)
absname = expr(self, e, abs_col=True)._expr
compound = True
cexprs.append(fname)
cname = ''.join([a if a in base62alp else '' for a in fname()])
compound = compound and has_groupby and self.datasource.rec not in self.group_node.referenced
cols.append(ColRef(f'{disp_varname}[{i}]', 'generic', self.out_table, 0, None, cname, i, compound=compound))
cols.append(ColRef(cname, expr.toCExpr(f'decays<decltype({absname})>')(0), self.out_table, 0, None, cname, i, compound=compound))
self.out_table.add_cols(cols, False)
cexpr += ')'
if has_groupby:
self.group_node.finalize(cexpr, disp_varname)
create_table(self, self.out_table)
self.group_node.finalize(cexprs, self.out_table)
else:
self.emit(f'auto {disp_varname} = {cexpr};')
create_table(self, self.out_table, cexpr = cexprs)
self.datasource.group_node = None
has_orderby = 'orderby' in node
@@ -122,12 +118,8 @@ class projection(ast_node):
self.emit_no_ln(f"{f'{disp_varname}:+' if flatten else ''}(")
if self.disp or has_orderby:
if len(self.projections) > 1:
self.emit_no_ln(f"{'+' if self.inv else ''}{disp_varname}")
else:
self.emit_no_ln(f'print({disp_varname});')
if flatten:
self.emit_no_ln(f'{disp_varname}')
self.emit(f'print(*{self.out_table.cxt_name});')
if has_orderby:
self.emit(f')[{orderby_node.view}]')
else:
engine/scan.py
+32 -2
View File
@@ -1,11 +1,41 @@
from xmlrpc.client import Boolean
from engine.ast import ColRef, TableInfo, View, ast_node
from engine.ast import ColRef, TableInfo, View, ast_node, Context
from engine.utils import base62uuid
from engine.expr import expr
class scan(ast_node):
name = 'scan'
def __init__(self, parent: "ast_node", node, size = None, context: Context = None):
self.type = type
self.size = size
super().__init__(parent, node, context)
def init(self, _):
self.datasource = self.context.datasource
self.start = ''
self.body = ''
self.end = '}'
self.filter = None
scan_vars = set(s.it_var for s in self.context.scans)
self.it_ver = 'i' + base62uuid(2)
while(self.it_ver in scan_vars):
self.it_ver = 'i' + base62uuid(6)
self.parent.context.scans.append(self)
def produce(self, node):
if type(node) is ColRef:
if self.size is None:
self.start += f'for (auto& {self.it_ver} : {node.reference()}) {{\n'
else:
self.start += f"for (uint32_t {self.it_ver} = 0; {self.it_ver} < {node.reference()}.size; ++{self.it_ver}){{\\n"
elif type(node) is str:
self.start+= f'for(auto& {self.it_ver} : {node}) {{\n'
else:
self.start += f"for (uint32_t {self.it_ver} = 0; {self.it_ver} < {self.size}; ++{self.it_ver}){{\n"
def add(self, stmt):
self.body+=stmt + '\n'
def finalize(self):
self.context.remove_scan(self, self.start + self.body + self.end)
class filter(ast_node):
name = 'filter'
def __init__(self, parent: "ast_node", node, materialize = False, context = None):