correct line_ending
This commit is contained in:
+10
-10
@@ -1,11 +1,11 @@
|
||||
# TODO:
|
||||
|
||||
## 1. double scans in projections
|
||||
- first for special aggrigations and singular columns
|
||||
- Then in group by node decide if we have special group by aggregations
|
||||
- If sp_gb_agg exists, the entire groupby aggregation is done in C plugin
|
||||
- If not, group by is done in SQL
|
||||
|
||||
## 2. ColRef supports multiple objects
|
||||
- A.a = B.b then in projection A.a B.b will refer to same projection
|
||||
# TODO:
|
||||
|
||||
## 1. double scans in projections
|
||||
- first for special aggrigations and singular columns
|
||||
- Then in group by node decide if we have special group by aggregations
|
||||
- If sp_gb_agg exists, the entire groupby aggregation is done in C plugin
|
||||
- If not, group by is done in SQL
|
||||
|
||||
## 2. ColRef supports multiple objects
|
||||
- A.a = B.b then in projection A.a B.b will refer to same projection
|
||||
- Colref::ProjEq(ColRef v) => this == v or v in this.proj_eqs
|
||||
+32
-32
@@ -1,32 +1,32 @@
|
||||
from reconstruct.ast import Context, ast_node
|
||||
saved_cxt = None
|
||||
|
||||
def initialize(cxt = None, keep = False):
|
||||
global saved_cxt
|
||||
if cxt is None or not keep or type(cxt) is not Context:
|
||||
if saved_cxt is None or not keep:
|
||||
cxt = Context()
|
||||
saved_cxt = cxt
|
||||
else:
|
||||
cxt = saved_cxt
|
||||
cxt.new()
|
||||
return cxt
|
||||
|
||||
def generate(ast, cxt):
|
||||
for k in ast.keys():
|
||||
if k in ast_node.types.keys():
|
||||
ast_node.types[k](None, ast, cxt)
|
||||
|
||||
def exec(stmts, cxt = None, keep = False):
|
||||
cxt = initialize(cxt, keep)
|
||||
stmts_stmts = stmts['stmts']
|
||||
if type(stmts_stmts) is list:
|
||||
for s in stmts_stmts:
|
||||
generate(s, cxt)
|
||||
else:
|
||||
generate(stmts_stmts, cxt)
|
||||
for q in cxt.queries:
|
||||
cxt.print(q.strip())
|
||||
return cxt
|
||||
|
||||
__all__ = ["initialize", "generate", "exec", "saved_cxt"]
|
||||
from reconstruct.ast import Context, ast_node
|
||||
saved_cxt = None
|
||||
|
||||
def initialize(cxt = None, keep = False):
|
||||
global saved_cxt
|
||||
if cxt is None or not keep or type(cxt) is not Context:
|
||||
if saved_cxt is None or not keep:
|
||||
cxt = Context()
|
||||
saved_cxt = cxt
|
||||
else:
|
||||
cxt = saved_cxt
|
||||
cxt.new()
|
||||
return cxt
|
||||
|
||||
def generate(ast, cxt):
|
||||
for k in ast.keys():
|
||||
if k in ast_node.types.keys():
|
||||
ast_node.types[k](None, ast, cxt)
|
||||
|
||||
def exec(stmts, cxt = None, keep = False):
|
||||
cxt = initialize(cxt, keep)
|
||||
stmts_stmts = stmts['stmts']
|
||||
if type(stmts_stmts) is list:
|
||||
for s in stmts_stmts:
|
||||
generate(s, cxt)
|
||||
else:
|
||||
generate(stmts_stmts, cxt)
|
||||
for q in cxt.queries:
|
||||
cxt.print(q.strip())
|
||||
return cxt
|
||||
|
||||
__all__ = ["initialize", "generate", "exec", "saved_cxt"]
|
||||
|
||||
+1048
-1048
File diff suppressed because it is too large
Load Diff
+337
-337
@@ -1,338 +1,338 @@
|
||||
from typing import Optional
|
||||
from reconstruct.ast import ast_node
|
||||
from reconstruct.storage import ColRef, Context
|
||||
from engine.types import *
|
||||
|
||||
# TODO: Decouple expr and upgrade architecture
|
||||
# C_CODE : get ccode/sql code?
|
||||
# projections : C/SQL/decltype string
|
||||
# orderby/joins/where : SQL only
|
||||
# assumption/groupby : C/sql
|
||||
# is_udfexpr: C only
|
||||
|
||||
class expr(ast_node):
|
||||
name='expr'
|
||||
@property
|
||||
def udf_decltypecall(self):
|
||||
return self._udf_decltypecall if self._udf_decltypecall else self.sql
|
||||
|
||||
@udf_decltypecall.setter
|
||||
def udf_decltypecall(self, val):
|
||||
self._udf_decltypecall = val
|
||||
|
||||
@property
|
||||
def need_decltypestr(self):
|
||||
return self._udf_decltypecall is not None
|
||||
|
||||
def __init__(self, parent, node, *, c_code = None, supress_undefined = False):
|
||||
from reconstruct.ast import projection, udf
|
||||
|
||||
self.type = None
|
||||
self.raw_col = None
|
||||
self.udf : Optional[udf] = None
|
||||
self.inside_agg = False
|
||||
self.is_special = False
|
||||
self.is_ColExpr = False
|
||||
self.is_recursive_call_inudf = False
|
||||
self.codlets : list = []
|
||||
self.codebuf : Optional[str] = None
|
||||
self._udf_decltypecall = None
|
||||
self.node = node
|
||||
self.supress_undefined = supress_undefined
|
||||
if(type(parent) is expr):
|
||||
self.inside_agg = parent.inside_agg
|
||||
self.is_udfexpr = parent.is_udfexpr
|
||||
self.is_agg_func = parent.is_agg_func
|
||||
self.root : expr = parent.root
|
||||
self.c_code = parent.c_code
|
||||
self.builtin_vars = parent.builtin_vars
|
||||
else:
|
||||
self.is_agg_func = False
|
||||
self.is_udfexpr = type(parent) is udf
|
||||
self.root : expr = self
|
||||
self.c_code = self.is_udfexpr or type(parent) is projection
|
||||
if self.is_udfexpr:
|
||||
self.udf : udf = parent
|
||||
self.builtin_vars = self.udf.builtin.keys()
|
||||
else:
|
||||
self.builtin_vars = []
|
||||
if type(c_code) is bool:
|
||||
self.c_code = c_code
|
||||
|
||||
self.udf_called = None
|
||||
self.cols_mentioned : Optional[set[ColRef]] = None
|
||||
ast_node.__init__(self, parent, node, None)
|
||||
|
||||
def init(self, _):
|
||||
from reconstruct.ast import projection
|
||||
parent = self.parent
|
||||
self.isvector = parent.isvector if type(parent) is expr else False
|
||||
self.is_compound = parent.is_compound if type(parent) is expr else False
|
||||
if type(parent) in [projection, expr]:
|
||||
self.datasource = parent.datasource
|
||||
else:
|
||||
self.datasource = self.context.datasource
|
||||
self.udf_map = parent.context.udf_map
|
||||
self.func_maps = {**builtin_func, **self.udf_map, **user_module_func}
|
||||
self.operators = {**builtin_operators, **self.udf_map, **user_module_func}
|
||||
|
||||
def produce(self, node):
|
||||
from engine.utils import enlist
|
||||
from reconstruct.ast import udf
|
||||
|
||||
if type(node) is dict:
|
||||
for key, val in node.items():
|
||||
if key in self.operators:
|
||||
if key in builtin_func:
|
||||
if self.is_agg_func:
|
||||
self.root.is_special = True # Nested Aggregation
|
||||
else:
|
||||
self.is_agg_func = True
|
||||
|
||||
op = self.operators[key]
|
||||
|
||||
val = enlist(val)
|
||||
exp_vals = [expr(self, v, c_code = self.c_code) for v in val]
|
||||
str_vals = [e.sql for e in exp_vals]
|
||||
type_vals = [e.type for e in exp_vals]
|
||||
try:
|
||||
self.type = op.return_type(*type_vals)
|
||||
except AttributeError as e:
|
||||
if type(self.root) is not udf:
|
||||
# TODO: do something when this is not an error
|
||||
# print(f'alert: {e}')
|
||||
pass
|
||||
self.type = AnyT
|
||||
|
||||
self.sql = op(self.c_code, *str_vals)
|
||||
special_func = [*self.context.udf_map.keys(), *self.context.module_map.keys(), "maxs", "mins", "avgs", "sums"]
|
||||
if key in special_func and not self.is_special:
|
||||
self.is_special = True
|
||||
if key in self.context.udf_map:
|
||||
self.root.udf_called = self.context.udf_map[key]
|
||||
if self.is_udfexpr and key == self.root.udf.name:
|
||||
self.root.is_recursive_call_inudf = True
|
||||
elif key in user_module_func.keys():
|
||||
udf.try_init_udf(self.context)
|
||||
# TODO: make udf_called a set!
|
||||
p = self.parent
|
||||
while type(p) is expr and not p.udf_called:
|
||||
p.udf_called = self.udf_called
|
||||
p = p.parent
|
||||
p = self.parent
|
||||
while type(p) is expr and not p.is_special:
|
||||
p.is_special = True
|
||||
p = p.parent
|
||||
|
||||
need_decltypestr = any([e.need_decltypestr for e in exp_vals])
|
||||
if need_decltypestr or (self.udf_called and type(op) is udf):
|
||||
decltypestr_vals = [e.udf_decltypecall for e in exp_vals]
|
||||
self.udf_decltypecall = op(self.c_code, *decltypestr_vals)
|
||||
|
||||
if self.udf_called and type(op) is udf:
|
||||
self.udf_decltypecall = op.decltypecall(self.c_code, *decltypestr_vals)
|
||||
|
||||
elif self.is_udfexpr:
|
||||
var_table = self.root.udf.var_table
|
||||
vec = key.split('.')
|
||||
_vars = [*var_table, *self.builtin_vars]
|
||||
def get_vname (node):
|
||||
if node in self.builtin_vars:
|
||||
self.root.udf.builtin[node].enabled = True
|
||||
self.builtin_var = node
|
||||
return node
|
||||
else:
|
||||
return var_table[node]
|
||||
if vec[0] not in _vars:
|
||||
# print(f'Use of undefined variable {vec[0]}')
|
||||
# TODO: do something when this is not an error
|
||||
pass
|
||||
else:
|
||||
vname = get_vname(vec[0])
|
||||
val = enlist(val)
|
||||
if(len(val) > 2):
|
||||
print('Warning: more than 2 indexes found for subvec operator.')
|
||||
ex = [expr(self, v, c_code = self.c_code) for v in val]
|
||||
idxs = ', '.join([e.sql for e in ex])
|
||||
self.sql = f'{vname}.subvec({idxs})'
|
||||
if any([e.need_decltypestr for e in ex]):
|
||||
self.udf_decltypecall = f'{vname}.subvec({[", ".join([e.udf_decltypecall for e in ex])]})'
|
||||
if key == 'get' and len(val) > 1:
|
||||
ex_vname = expr(self, val[0], c_code=self.c_code)
|
||||
self.sql = f'{ex_vname.sql}[{expr(self, val[1], c_code=self.c_code).sql}]'
|
||||
if hasattr(ex_vname, 'builtin_var'):
|
||||
if not hasattr(self, 'builtin_var'):
|
||||
self.builtin_var = []
|
||||
self.builtin_var = [*self.builtin_var, *ex_vname.builtin_var]
|
||||
self.udf_decltypecall = ex_vname.sql
|
||||
else:
|
||||
print(f'Undefined expr: {key}{val}')
|
||||
|
||||
elif type(node) is str:
|
||||
if self.is_udfexpr:
|
||||
curr_udf : udf = self.root.udf
|
||||
var_table = curr_udf.var_table
|
||||
split = node.split('.')
|
||||
if split[0] in var_table:
|
||||
varname = var_table[split[0]]
|
||||
if curr_udf.agg and varname in curr_udf.vecs:
|
||||
if len(split) > 1:
|
||||
if split[1] == 'vec':
|
||||
self.sql += varname
|
||||
elif split[1] == 'len':
|
||||
self.sql += f'{varname}.size'
|
||||
else:
|
||||
print(f'no member {split[1]} in object {varname}')
|
||||
else:
|
||||
self.sql += f'{varname}[{curr_udf.idx_var}]'
|
||||
else:
|
||||
self.sql += varname
|
||||
elif self.supress_undefined or split[0] in self.builtin_vars:
|
||||
self.sql += node
|
||||
if split[0] in self.builtin_vars:
|
||||
curr_udf.builtin[split[0]].enabled = True
|
||||
self.builtin_var = split[0]
|
||||
else:
|
||||
print(f'Undefined varname: {split[0]}')
|
||||
|
||||
|
||||
# get the column from the datasource in SQL context
|
||||
else:
|
||||
p = self.parent
|
||||
while type(p) is expr and not p.isvector:
|
||||
p.isvector = True
|
||||
p = p.parent
|
||||
if self.datasource is not None:
|
||||
self.raw_col = self.datasource.parse_col_names(node)
|
||||
self.raw_col = self.raw_col if type(self.raw_col) is ColRef else None
|
||||
if self.raw_col is not None:
|
||||
self.is_ColExpr = True
|
||||
self.sql = self.raw_col.name
|
||||
self.type = self.raw_col.type
|
||||
else:
|
||||
self.sql = node
|
||||
self.type = StrT
|
||||
if self.c_code and self.datasource is not None:
|
||||
self.sql = f'{{y(\"{self.sql}\")}}'
|
||||
elif type(node) is bool:
|
||||
self.type = BoolT
|
||||
if self.c_code:
|
||||
self.sql = '1' if node else '0'
|
||||
else:
|
||||
self.sql = 'TRUE' if node else 'FALSE'
|
||||
else:
|
||||
self.sql = f'{node}'
|
||||
if type(node) is int:
|
||||
if (node >= 2**63 - 1 or node <= -2**63):
|
||||
self.type = LongT
|
||||
else:
|
||||
self.type = IntT
|
||||
elif type(node) is float:
|
||||
self.type = DoubleT
|
||||
|
||||
def finalize(self, override = False):
|
||||
from reconstruct.ast import udf
|
||||
if self.codebuf is None or override:
|
||||
self.codebuf = ''
|
||||
for c in self.codlets:
|
||||
if type(c) is str:
|
||||
self.codebuf += c
|
||||
elif type(c) is udf:
|
||||
self.codebuf += c()
|
||||
elif type(c) is expr:
|
||||
self.codebuf += c.finalize(override=override)
|
||||
return self.codebuf
|
||||
|
||||
def __str__(self):
|
||||
return self.sql
|
||||
def __repr__(self):
|
||||
return self.__str__()
|
||||
|
||||
# builtins is readonly, so it's okay to set default value as an object
|
||||
# eval is only called at root expr.
|
||||
def eval(self, c_code = None, y = lambda t: t, materialize_builtin = False, _decltypestr = False, *, gettype = False):
|
||||
assert(self.is_root)
|
||||
def call(decltypestr = False) -> str:
|
||||
nonlocal c_code, y, materialize_builtin
|
||||
if self.udf_called is not None:
|
||||
loc = locals()
|
||||
builtin_vars = self.udf_called.builtin_used
|
||||
for b in self.udf_called.builtin_var.all:
|
||||
exec(f'loc["{b}"] = lambda: "{{{b}()}}"')
|
||||
if builtin_vars:
|
||||
if type(materialize_builtin) is dict:
|
||||
for b in builtin_vars:
|
||||
exec(f'loc["{b}"] = lambda: "{materialize_builtin[b]}"')
|
||||
elif self.is_recursive_call_inudf:
|
||||
for b in builtin_vars:
|
||||
exec(f'loc["{b}"] = lambda : "{b}"')
|
||||
|
||||
x = self.c_code if c_code is None else c_code
|
||||
if decltypestr:
|
||||
return eval('f\'' + self.udf_decltypecall + '\'')
|
||||
return eval('f\'' + self.sql + '\'')
|
||||
if self.is_recursive_call_inudf or (self.need_decltypestr and self.is_udfexpr) or gettype:
|
||||
return call
|
||||
else:
|
||||
return call(_decltypestr)
|
||||
|
||||
@property
|
||||
def is_root(self):
|
||||
return self.root == self
|
||||
|
||||
|
||||
# For UDFs: first check if agg variable is used as vector
|
||||
# if not, then check if its length is used
|
||||
class fastscan(expr):
|
||||
name = 'fastscan'
|
||||
|
||||
def init(self, _):
|
||||
self.vec_vars = set()
|
||||
self.requested_lens = set()
|
||||
super().init(self, _)
|
||||
|
||||
def process(self, key : str):
|
||||
segs = key.split('.')
|
||||
var_table = self.root.udf.var_table
|
||||
if segs[0] in var_table and len(segs) > 1:
|
||||
if segs[1] == 'vec':
|
||||
self.vec_vars.add(segs[0])
|
||||
elif segs[1] == 'len':
|
||||
self.requested_lens.add(segs[0])
|
||||
|
||||
def produce(self, node):
|
||||
from engine.utils import enlist
|
||||
if type(node) is dict:
|
||||
for key, val in node.items():
|
||||
if key in self.operators:
|
||||
val = enlist(val)
|
||||
elif self.is_udfexpr:
|
||||
self.process(key)
|
||||
[fastscan(self, v, c_code = self.c_code) for v in val]
|
||||
|
||||
elif type(node) is str:
|
||||
self.process(node)
|
||||
|
||||
|
||||
class getrefs(expr):
|
||||
name = 'getrefs'
|
||||
|
||||
def init(self, _):
|
||||
self.datasource.rec = set()
|
||||
self.rec = None
|
||||
|
||||
def produce(self, node):
|
||||
from engine.utils import enlist
|
||||
if type(node) is dict:
|
||||
for key, val in node.items():
|
||||
if key in self.operators:
|
||||
val = enlist(val)
|
||||
[getrefs(self, v, c_code = self.c_code) for v in val]
|
||||
|
||||
elif type(node) is str:
|
||||
self.datasource.parse_col_names(node)
|
||||
|
||||
def consume(self, _):
|
||||
if self.root == self:
|
||||
self.rec = self.datasource.rec
|
||||
from typing import Optional
|
||||
from reconstruct.ast import ast_node
|
||||
from reconstruct.storage import ColRef, Context
|
||||
from engine.types import *
|
||||
|
||||
# TODO: Decouple expr and upgrade architecture
|
||||
# C_CODE : get ccode/sql code?
|
||||
# projections : C/SQL/decltype string
|
||||
# orderby/joins/where : SQL only
|
||||
# assumption/groupby : C/sql
|
||||
# is_udfexpr: C only
|
||||
|
||||
class expr(ast_node):
|
||||
name='expr'
|
||||
@property
|
||||
def udf_decltypecall(self):
|
||||
return self._udf_decltypecall if self._udf_decltypecall else self.sql
|
||||
|
||||
@udf_decltypecall.setter
|
||||
def udf_decltypecall(self, val):
|
||||
self._udf_decltypecall = val
|
||||
|
||||
@property
|
||||
def need_decltypestr(self):
|
||||
return self._udf_decltypecall is not None
|
||||
|
||||
def __init__(self, parent, node, *, c_code = None, supress_undefined = False):
|
||||
from reconstruct.ast import projection, udf
|
||||
|
||||
self.type = None
|
||||
self.raw_col = None
|
||||
self.udf : Optional[udf] = None
|
||||
self.inside_agg = False
|
||||
self.is_special = False
|
||||
self.is_ColExpr = False
|
||||
self.is_recursive_call_inudf = False
|
||||
self.codlets : list = []
|
||||
self.codebuf : Optional[str] = None
|
||||
self._udf_decltypecall = None
|
||||
self.node = node
|
||||
self.supress_undefined = supress_undefined
|
||||
if(type(parent) is expr):
|
||||
self.inside_agg = parent.inside_agg
|
||||
self.is_udfexpr = parent.is_udfexpr
|
||||
self.is_agg_func = parent.is_agg_func
|
||||
self.root : expr = parent.root
|
||||
self.c_code = parent.c_code
|
||||
self.builtin_vars = parent.builtin_vars
|
||||
else:
|
||||
self.is_agg_func = False
|
||||
self.is_udfexpr = type(parent) is udf
|
||||
self.root : expr = self
|
||||
self.c_code = self.is_udfexpr or type(parent) is projection
|
||||
if self.is_udfexpr:
|
||||
self.udf : udf = parent
|
||||
self.builtin_vars = self.udf.builtin.keys()
|
||||
else:
|
||||
self.builtin_vars = []
|
||||
if type(c_code) is bool:
|
||||
self.c_code = c_code
|
||||
|
||||
self.udf_called = None
|
||||
self.cols_mentioned : Optional[set[ColRef]] = None
|
||||
ast_node.__init__(self, parent, node, None)
|
||||
|
||||
def init(self, _):
|
||||
from reconstruct.ast import projection
|
||||
parent = self.parent
|
||||
self.isvector = parent.isvector if type(parent) is expr else False
|
||||
self.is_compound = parent.is_compound if type(parent) is expr else False
|
||||
if type(parent) in [projection, expr]:
|
||||
self.datasource = parent.datasource
|
||||
else:
|
||||
self.datasource = self.context.datasource
|
||||
self.udf_map = parent.context.udf_map
|
||||
self.func_maps = {**builtin_func, **self.udf_map, **user_module_func}
|
||||
self.operators = {**builtin_operators, **self.udf_map, **user_module_func}
|
||||
|
||||
def produce(self, node):
|
||||
from engine.utils import enlist
|
||||
from reconstruct.ast import udf
|
||||
|
||||
if type(node) is dict:
|
||||
for key, val in node.items():
|
||||
if key in self.operators:
|
||||
if key in builtin_func:
|
||||
if self.is_agg_func:
|
||||
self.root.is_special = True # Nested Aggregation
|
||||
else:
|
||||
self.is_agg_func = True
|
||||
|
||||
op = self.operators[key]
|
||||
|
||||
val = enlist(val)
|
||||
exp_vals = [expr(self, v, c_code = self.c_code) for v in val]
|
||||
str_vals = [e.sql for e in exp_vals]
|
||||
type_vals = [e.type for e in exp_vals]
|
||||
try:
|
||||
self.type = op.return_type(*type_vals)
|
||||
except AttributeError as e:
|
||||
if type(self.root) is not udf:
|
||||
# TODO: do something when this is not an error
|
||||
# print(f'alert: {e}')
|
||||
pass
|
||||
self.type = AnyT
|
||||
|
||||
self.sql = op(self.c_code, *str_vals)
|
||||
special_func = [*self.context.udf_map.keys(), *self.context.module_map.keys(), "maxs", "mins", "avgs", "sums"]
|
||||
if key in special_func and not self.is_special:
|
||||
self.is_special = True
|
||||
if key in self.context.udf_map:
|
||||
self.root.udf_called = self.context.udf_map[key]
|
||||
if self.is_udfexpr and key == self.root.udf.name:
|
||||
self.root.is_recursive_call_inudf = True
|
||||
elif key in user_module_func.keys():
|
||||
udf.try_init_udf(self.context)
|
||||
# TODO: make udf_called a set!
|
||||
p = self.parent
|
||||
while type(p) is expr and not p.udf_called:
|
||||
p.udf_called = self.udf_called
|
||||
p = p.parent
|
||||
p = self.parent
|
||||
while type(p) is expr and not p.is_special:
|
||||
p.is_special = True
|
||||
p = p.parent
|
||||
|
||||
need_decltypestr = any([e.need_decltypestr for e in exp_vals])
|
||||
if need_decltypestr or (self.udf_called and type(op) is udf):
|
||||
decltypestr_vals = [e.udf_decltypecall for e in exp_vals]
|
||||
self.udf_decltypecall = op(self.c_code, *decltypestr_vals)
|
||||
|
||||
if self.udf_called and type(op) is udf:
|
||||
self.udf_decltypecall = op.decltypecall(self.c_code, *decltypestr_vals)
|
||||
|
||||
elif self.is_udfexpr:
|
||||
var_table = self.root.udf.var_table
|
||||
vec = key.split('.')
|
||||
_vars = [*var_table, *self.builtin_vars]
|
||||
def get_vname (node):
|
||||
if node in self.builtin_vars:
|
||||
self.root.udf.builtin[node].enabled = True
|
||||
self.builtin_var = node
|
||||
return node
|
||||
else:
|
||||
return var_table[node]
|
||||
if vec[0] not in _vars:
|
||||
# print(f'Use of undefined variable {vec[0]}')
|
||||
# TODO: do something when this is not an error
|
||||
pass
|
||||
else:
|
||||
vname = get_vname(vec[0])
|
||||
val = enlist(val)
|
||||
if(len(val) > 2):
|
||||
print('Warning: more than 2 indexes found for subvec operator.')
|
||||
ex = [expr(self, v, c_code = self.c_code) for v in val]
|
||||
idxs = ', '.join([e.sql for e in ex])
|
||||
self.sql = f'{vname}.subvec({idxs})'
|
||||
if any([e.need_decltypestr for e in ex]):
|
||||
self.udf_decltypecall = f'{vname}.subvec({[", ".join([e.udf_decltypecall for e in ex])]})'
|
||||
if key == 'get' and len(val) > 1:
|
||||
ex_vname = expr(self, val[0], c_code=self.c_code)
|
||||
self.sql = f'{ex_vname.sql}[{expr(self, val[1], c_code=self.c_code).sql}]'
|
||||
if hasattr(ex_vname, 'builtin_var'):
|
||||
if not hasattr(self, 'builtin_var'):
|
||||
self.builtin_var = []
|
||||
self.builtin_var = [*self.builtin_var, *ex_vname.builtin_var]
|
||||
self.udf_decltypecall = ex_vname.sql
|
||||
else:
|
||||
print(f'Undefined expr: {key}{val}')
|
||||
|
||||
elif type(node) is str:
|
||||
if self.is_udfexpr:
|
||||
curr_udf : udf = self.root.udf
|
||||
var_table = curr_udf.var_table
|
||||
split = node.split('.')
|
||||
if split[0] in var_table:
|
||||
varname = var_table[split[0]]
|
||||
if curr_udf.agg and varname in curr_udf.vecs:
|
||||
if len(split) > 1:
|
||||
if split[1] == 'vec':
|
||||
self.sql += varname
|
||||
elif split[1] == 'len':
|
||||
self.sql += f'{varname}.size'
|
||||
else:
|
||||
print(f'no member {split[1]} in object {varname}')
|
||||
else:
|
||||
self.sql += f'{varname}[{curr_udf.idx_var}]'
|
||||
else:
|
||||
self.sql += varname
|
||||
elif self.supress_undefined or split[0] in self.builtin_vars:
|
||||
self.sql += node
|
||||
if split[0] in self.builtin_vars:
|
||||
curr_udf.builtin[split[0]].enabled = True
|
||||
self.builtin_var = split[0]
|
||||
else:
|
||||
print(f'Undefined varname: {split[0]}')
|
||||
|
||||
|
||||
# get the column from the datasource in SQL context
|
||||
else:
|
||||
p = self.parent
|
||||
while type(p) is expr and not p.isvector:
|
||||
p.isvector = True
|
||||
p = p.parent
|
||||
if self.datasource is not None:
|
||||
self.raw_col = self.datasource.parse_col_names(node)
|
||||
self.raw_col = self.raw_col if type(self.raw_col) is ColRef else None
|
||||
if self.raw_col is not None:
|
||||
self.is_ColExpr = True
|
||||
self.sql = self.raw_col.name
|
||||
self.type = self.raw_col.type
|
||||
else:
|
||||
self.sql = node
|
||||
self.type = StrT
|
||||
if self.c_code and self.datasource is not None:
|
||||
self.sql = f'{{y(\"{self.sql}\")}}'
|
||||
elif type(node) is bool:
|
||||
self.type = BoolT
|
||||
if self.c_code:
|
||||
self.sql = '1' if node else '0'
|
||||
else:
|
||||
self.sql = 'TRUE' if node else 'FALSE'
|
||||
else:
|
||||
self.sql = f'{node}'
|
||||
if type(node) is int:
|
||||
if (node >= 2**63 - 1 or node <= -2**63):
|
||||
self.type = LongT
|
||||
else:
|
||||
self.type = IntT
|
||||
elif type(node) is float:
|
||||
self.type = DoubleT
|
||||
|
||||
def finalize(self, override = False):
|
||||
from reconstruct.ast import udf
|
||||
if self.codebuf is None or override:
|
||||
self.codebuf = ''
|
||||
for c in self.codlets:
|
||||
if type(c) is str:
|
||||
self.codebuf += c
|
||||
elif type(c) is udf:
|
||||
self.codebuf += c()
|
||||
elif type(c) is expr:
|
||||
self.codebuf += c.finalize(override=override)
|
||||
return self.codebuf
|
||||
|
||||
def __str__(self):
|
||||
return self.sql
|
||||
def __repr__(self):
|
||||
return self.__str__()
|
||||
|
||||
# builtins is readonly, so it's okay to set default value as an object
|
||||
# eval is only called at root expr.
|
||||
def eval(self, c_code = None, y = lambda t: t, materialize_builtin = False, _decltypestr = False, *, gettype = False):
|
||||
assert(self.is_root)
|
||||
def call(decltypestr = False) -> str:
|
||||
nonlocal c_code, y, materialize_builtin
|
||||
if self.udf_called is not None:
|
||||
loc = locals()
|
||||
builtin_vars = self.udf_called.builtin_used
|
||||
for b in self.udf_called.builtin_var.all:
|
||||
exec(f'loc["{b}"] = lambda: "{{{b}()}}"')
|
||||
if builtin_vars:
|
||||
if type(materialize_builtin) is dict:
|
||||
for b in builtin_vars:
|
||||
exec(f'loc["{b}"] = lambda: "{materialize_builtin[b]}"')
|
||||
elif self.is_recursive_call_inudf:
|
||||
for b in builtin_vars:
|
||||
exec(f'loc["{b}"] = lambda : "{b}"')
|
||||
|
||||
x = self.c_code if c_code is None else c_code
|
||||
if decltypestr:
|
||||
return eval('f\'' + self.udf_decltypecall + '\'')
|
||||
return eval('f\'' + self.sql + '\'')
|
||||
if self.is_recursive_call_inudf or (self.need_decltypestr and self.is_udfexpr) or gettype:
|
||||
return call
|
||||
else:
|
||||
return call(_decltypestr)
|
||||
|
||||
@property
|
||||
def is_root(self):
|
||||
return self.root == self
|
||||
|
||||
|
||||
# For UDFs: first check if agg variable is used as vector
|
||||
# if not, then check if its length is used
|
||||
class fastscan(expr):
|
||||
name = 'fastscan'
|
||||
|
||||
def init(self, _):
|
||||
self.vec_vars = set()
|
||||
self.requested_lens = set()
|
||||
super().init(self, _)
|
||||
|
||||
def process(self, key : str):
|
||||
segs = key.split('.')
|
||||
var_table = self.root.udf.var_table
|
||||
if segs[0] in var_table and len(segs) > 1:
|
||||
if segs[1] == 'vec':
|
||||
self.vec_vars.add(segs[0])
|
||||
elif segs[1] == 'len':
|
||||
self.requested_lens.add(segs[0])
|
||||
|
||||
def produce(self, node):
|
||||
from engine.utils import enlist
|
||||
if type(node) is dict:
|
||||
for key, val in node.items():
|
||||
if key in self.operators:
|
||||
val = enlist(val)
|
||||
elif self.is_udfexpr:
|
||||
self.process(key)
|
||||
[fastscan(self, v, c_code = self.c_code) for v in val]
|
||||
|
||||
elif type(node) is str:
|
||||
self.process(node)
|
||||
|
||||
|
||||
class getrefs(expr):
|
||||
name = 'getrefs'
|
||||
|
||||
def init(self, _):
|
||||
self.datasource.rec = set()
|
||||
self.rec = None
|
||||
|
||||
def produce(self, node):
|
||||
from engine.utils import enlist
|
||||
if type(node) is dict:
|
||||
for key, val in node.items():
|
||||
if key in self.operators:
|
||||
val = enlist(val)
|
||||
[getrefs(self, v, c_code = self.c_code) for v in val]
|
||||
|
||||
elif type(node) is str:
|
||||
self.datasource.parse_col_names(node)
|
||||
|
||||
def consume(self, _):
|
||||
if self.root == self:
|
||||
self.rec = self.datasource.rec
|
||||
self.datasource.rec = None
|
||||
+203
-203
@@ -1,203 +1,203 @@
|
||||
from engine.types import *
|
||||
from engine.utils import enlist
|
||||
from typing import List, Dict, Set
|
||||
|
||||
class ColRef:
|
||||
def __init__(self, _ty, cobj, table:'TableInfo', name, id, compound = False, _ty_args = None):
|
||||
self.type : Types = AnyT
|
||||
if type(_ty) is str:
|
||||
self.type = builtin_types[_ty.lower()]
|
||||
if _ty_args:
|
||||
self.type = self.type(enlist(_ty_args))
|
||||
elif type(_ty) is Types:
|
||||
self.type = _ty
|
||||
self.cobj = cobj
|
||||
self.table = table
|
||||
self.name = name
|
||||
self.alias = set()
|
||||
self.id = id # position in table
|
||||
self.compound = compound # compound field (list as a field)
|
||||
# e.g. order by, group by, filter by expressions
|
||||
|
||||
self.__arr__ = (_ty, cobj, table, name, id)
|
||||
def __getitem__(self, key):
|
||||
if type(key) is str:
|
||||
return getattr(self, key)
|
||||
else:
|
||||
return self.__arr__[key]
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
self.__arr__[key] = value
|
||||
|
||||
class TableInfo:
|
||||
def __init__(self, table_name, cols, cxt:'Context'):
|
||||
# statics
|
||||
self.table_name : str = table_name
|
||||
self.contextname_cpp : str = ''
|
||||
self.alias : Set[str] = set([table_name])
|
||||
self.columns_byname : Dict[str, ColRef] = dict() # column_name, type
|
||||
self.columns : List[ColRef] = []
|
||||
self.cxt = cxt
|
||||
# keep track of temp vars
|
||||
self.rec = None
|
||||
self.add_cols(cols)
|
||||
# runtime
|
||||
self.order = [] # assumptions
|
||||
|
||||
cxt.tables_byname[self.table_name] = self # construct reverse map
|
||||
|
||||
def add_cols(self, cols, new = True):
|
||||
for i, c in enumerate(cols):
|
||||
self.add_col(c, new, i)
|
||||
|
||||
def add_col(self, c, new = True, i = 0):
|
||||
_ty = c['type']
|
||||
_ty_args = None
|
||||
if type(_ty) is dict:
|
||||
_ty_val = list(_ty.keys())[0]
|
||||
_ty_args = _ty[_ty_val]
|
||||
_ty = _ty_val
|
||||
if new:
|
||||
col_object = ColRef(_ty, c, self, c['name'], len(self.columns), _ty_args = _ty_args)
|
||||
else:
|
||||
col_object = c
|
||||
c.table = self
|
||||
self.columns_byname[c['name']] = col_object
|
||||
self.columns.append(col_object)
|
||||
|
||||
def add_alias(self, alias):
|
||||
if alias in self.cxt.tables_byname.keys():
|
||||
print("Error: table alias already exists")
|
||||
return
|
||||
self.cxt.tables_byname[alias] = self
|
||||
self.alias.add(alias)
|
||||
|
||||
def parse_col_names(self, colExpr) -> ColRef:
|
||||
parsedColExpr = colExpr.split('.')
|
||||
if len(parsedColExpr) <= 1:
|
||||
col = self.columns_byname[colExpr]
|
||||
if type(self.rec) is set:
|
||||
self.rec.add(col)
|
||||
return col
|
||||
else:
|
||||
datasource = self.cxt.tables_byname[parsedColExpr[0]]
|
||||
if datasource is None:
|
||||
raise ValueError(f'Table name/alias not defined{parsedColExpr[0]}')
|
||||
else:
|
||||
return datasource.parse_col_names(parsedColExpr[1])
|
||||
|
||||
|
||||
class Context:
|
||||
def new(self):
|
||||
self.headers = set(['\"./server/libaquery.h\"',
|
||||
'\"./server/monetdb_conn.h\"'])
|
||||
|
||||
self.ccode = ''
|
||||
self.sql = ''
|
||||
self.finalized = False
|
||||
self.udf = None
|
||||
self.scans = []
|
||||
self.procs = []
|
||||
self.queries = []
|
||||
self.module_init_loc = 0
|
||||
|
||||
def __init__(self):
|
||||
self.tables_byname = dict()
|
||||
self.col_byname = dict()
|
||||
self.tables = []
|
||||
self.cols = []
|
||||
self.datasource = None
|
||||
self.module_stubs = ''
|
||||
self.module_map = {}
|
||||
self.udf_map = dict()
|
||||
self.udf_agg_map = dict()
|
||||
self.use_columnstore = False
|
||||
self.print = print
|
||||
self.has_dll = False
|
||||
self.dialect = 'MonetDB'
|
||||
self.is_msvc = False
|
||||
self.have_hge = False
|
||||
self.Error = lambda *args: print(*args)
|
||||
self.Info = lambda *_: None
|
||||
|
||||
def emit(self, sql:str):
|
||||
self.sql += sql + ' '
|
||||
def emitc(self, c:str):
|
||||
self.ccode += c + '\n'
|
||||
def add_table(self, table_name, cols):
|
||||
tbl = TableInfo(table_name, cols, self)
|
||||
self.tables.append(tbl)
|
||||
return tbl
|
||||
def remove_scan(self, scan, str_scan):
|
||||
self.emitc(str_scan)
|
||||
self.scans.remove(scan)
|
||||
|
||||
function_deco = '__AQEXPORT__(int) '
|
||||
function_head = ('(Context* cxt) {\n' +
|
||||
'\tusing namespace std;\n' +
|
||||
'\tusing namespace types;\n' +
|
||||
'\tauto server = static_cast<Server*>(cxt->alt_server);\n')
|
||||
|
||||
udf_head = ('#pragma once\n'
|
||||
'#include \"./server/libaquery.h\"\n'
|
||||
'#include \"./server/aggregations.h\"\n\n'
|
||||
)
|
||||
|
||||
def get_init_func(self):
|
||||
if not self.module_map:
|
||||
return ''
|
||||
ret = '__AQEXPORT__(void) __builtin_init_user_module(Context* cxt){\n'
|
||||
for fname in self.module_map.keys():
|
||||
ret += f'{fname} = (decltype({fname}))(cxt->get_module_function("{fname}"));\n'
|
||||
self.queries.insert(self.module_init_loc, 'P__builtin_init_user_module')
|
||||
return ret + '}\n'
|
||||
|
||||
def sql_begin(self):
|
||||
self.sql = ''
|
||||
|
||||
def sql_end(self):
|
||||
# eliminate empty queries
|
||||
s = self.sql.strip()
|
||||
while(s and s[-1] == ';'):
|
||||
s = s[:-1].strip()
|
||||
if s and s.lower() != 'select':
|
||||
self.queries.append('Q' + self.sql)
|
||||
self.sql = ''
|
||||
|
||||
def postproc_begin(self, proc_name: str):
|
||||
self.ccode = self.function_deco + proc_name + self.function_head
|
||||
|
||||
def postproc_end(self, proc_name: str):
|
||||
self.procs.append(self.ccode + 'return 0;\n}')
|
||||
self.ccode = ''
|
||||
self.queries.append('P' + proc_name)
|
||||
|
||||
def finalize_udf(self):
|
||||
if self.udf is not None:
|
||||
return (Context.udf_head
|
||||
+ self.module_stubs
|
||||
+ self.get_init_func()
|
||||
+ self.udf
|
||||
)
|
||||
else:
|
||||
return None
|
||||
|
||||
def finalize(self):
|
||||
from aquery_config import build_driver, os_platform
|
||||
if not self.finalized:
|
||||
headers = ''
|
||||
if build_driver == 'MSBuild':
|
||||
headers ='#include \"./server/pch.hpp\"\n'
|
||||
|
||||
for h in self.headers:
|
||||
if h[0] != '"':
|
||||
headers += '#include <' + h + '>\n'
|
||||
else:
|
||||
headers += '#include ' + h + '\n'
|
||||
if os_platform == 'win':
|
||||
headers += '#undef max\n'
|
||||
headers += '#undef min\n'
|
||||
|
||||
self.ccode = headers + '\n'.join(self.procs)
|
||||
self.headers = set()
|
||||
return self.ccode
|
||||
from engine.types import *
|
||||
from engine.utils import enlist
|
||||
from typing import List, Dict, Set
|
||||
|
||||
class ColRef:
|
||||
def __init__(self, _ty, cobj, table:'TableInfo', name, id, compound = False, _ty_args = None):
|
||||
self.type : Types = AnyT
|
||||
if type(_ty) is str:
|
||||
self.type = builtin_types[_ty.lower()]
|
||||
if _ty_args:
|
||||
self.type = self.type(enlist(_ty_args))
|
||||
elif type(_ty) is Types:
|
||||
self.type = _ty
|
||||
self.cobj = cobj
|
||||
self.table = table
|
||||
self.name = name
|
||||
self.alias = set()
|
||||
self.id = id # position in table
|
||||
self.compound = compound # compound field (list as a field)
|
||||
# e.g. order by, group by, filter by expressions
|
||||
|
||||
self.__arr__ = (_ty, cobj, table, name, id)
|
||||
def __getitem__(self, key):
|
||||
if type(key) is str:
|
||||
return getattr(self, key)
|
||||
else:
|
||||
return self.__arr__[key]
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
self.__arr__[key] = value
|
||||
|
||||
class TableInfo:
|
||||
def __init__(self, table_name, cols, cxt:'Context'):
|
||||
# statics
|
||||
self.table_name : str = table_name
|
||||
self.contextname_cpp : str = ''
|
||||
self.alias : Set[str] = set([table_name])
|
||||
self.columns_byname : Dict[str, ColRef] = dict() # column_name, type
|
||||
self.columns : List[ColRef] = []
|
||||
self.cxt = cxt
|
||||
# keep track of temp vars
|
||||
self.rec = None
|
||||
self.add_cols(cols)
|
||||
# runtime
|
||||
self.order = [] # assumptions
|
||||
|
||||
cxt.tables_byname[self.table_name] = self # construct reverse map
|
||||
|
||||
def add_cols(self, cols, new = True):
|
||||
for i, c in enumerate(cols):
|
||||
self.add_col(c, new, i)
|
||||
|
||||
def add_col(self, c, new = True, i = 0):
|
||||
_ty = c['type']
|
||||
_ty_args = None
|
||||
if type(_ty) is dict:
|
||||
_ty_val = list(_ty.keys())[0]
|
||||
_ty_args = _ty[_ty_val]
|
||||
_ty = _ty_val
|
||||
if new:
|
||||
col_object = ColRef(_ty, c, self, c['name'], len(self.columns), _ty_args = _ty_args)
|
||||
else:
|
||||
col_object = c
|
||||
c.table = self
|
||||
self.columns_byname[c['name']] = col_object
|
||||
self.columns.append(col_object)
|
||||
|
||||
def add_alias(self, alias):
|
||||
if alias in self.cxt.tables_byname.keys():
|
||||
print("Error: table alias already exists")
|
||||
return
|
||||
self.cxt.tables_byname[alias] = self
|
||||
self.alias.add(alias)
|
||||
|
||||
def parse_col_names(self, colExpr) -> ColRef:
|
||||
parsedColExpr = colExpr.split('.')
|
||||
if len(parsedColExpr) <= 1:
|
||||
col = self.columns_byname[colExpr]
|
||||
if type(self.rec) is set:
|
||||
self.rec.add(col)
|
||||
return col
|
||||
else:
|
||||
datasource = self.cxt.tables_byname[parsedColExpr[0]]
|
||||
if datasource is None:
|
||||
raise ValueError(f'Table name/alias not defined{parsedColExpr[0]}')
|
||||
else:
|
||||
return datasource.parse_col_names(parsedColExpr[1])
|
||||
|
||||
|
||||
class Context:
|
||||
def new(self):
|
||||
self.headers = set(['\"./server/libaquery.h\"',
|
||||
'\"./server/monetdb_conn.h\"'])
|
||||
|
||||
self.ccode = ''
|
||||
self.sql = ''
|
||||
self.finalized = False
|
||||
self.udf = None
|
||||
self.scans = []
|
||||
self.procs = []
|
||||
self.queries = []
|
||||
self.module_init_loc = 0
|
||||
|
||||
def __init__(self):
|
||||
self.tables_byname = dict()
|
||||
self.col_byname = dict()
|
||||
self.tables = []
|
||||
self.cols = []
|
||||
self.datasource = None
|
||||
self.module_stubs = ''
|
||||
self.module_map = {}
|
||||
self.udf_map = dict()
|
||||
self.udf_agg_map = dict()
|
||||
self.use_columnstore = False
|
||||
self.print = print
|
||||
self.has_dll = False
|
||||
self.dialect = 'MonetDB'
|
||||
self.is_msvc = False
|
||||
self.have_hge = False
|
||||
self.Error = lambda *args: print(*args)
|
||||
self.Info = lambda *_: None
|
||||
|
||||
def emit(self, sql:str):
|
||||
self.sql += sql + ' '
|
||||
def emitc(self, c:str):
|
||||
self.ccode += c + '\n'
|
||||
def add_table(self, table_name, cols):
|
||||
tbl = TableInfo(table_name, cols, self)
|
||||
self.tables.append(tbl)
|
||||
return tbl
|
||||
def remove_scan(self, scan, str_scan):
|
||||
self.emitc(str_scan)
|
||||
self.scans.remove(scan)
|
||||
|
||||
function_deco = '__AQEXPORT__(int) '
|
||||
function_head = ('(Context* cxt) {\n' +
|
||||
'\tusing namespace std;\n' +
|
||||
'\tusing namespace types;\n' +
|
||||
'\tauto server = static_cast<Server*>(cxt->alt_server);\n')
|
||||
|
||||
udf_head = ('#pragma once\n'
|
||||
'#include \"./server/libaquery.h\"\n'
|
||||
'#include \"./server/aggregations.h\"\n\n'
|
||||
)
|
||||
|
||||
def get_init_func(self):
|
||||
if not self.module_map:
|
||||
return ''
|
||||
ret = '__AQEXPORT__(void) __builtin_init_user_module(Context* cxt){\n'
|
||||
for fname in self.module_map.keys():
|
||||
ret += f'{fname} = (decltype({fname}))(cxt->get_module_function("{fname}"));\n'
|
||||
self.queries.insert(self.module_init_loc, 'P__builtin_init_user_module')
|
||||
return ret + '}\n'
|
||||
|
||||
def sql_begin(self):
|
||||
self.sql = ''
|
||||
|
||||
def sql_end(self):
|
||||
# eliminate empty queries
|
||||
s = self.sql.strip()
|
||||
while(s and s[-1] == ';'):
|
||||
s = s[:-1].strip()
|
||||
if s and s.lower() != 'select':
|
||||
self.queries.append('Q' + self.sql)
|
||||
self.sql = ''
|
||||
|
||||
def postproc_begin(self, proc_name: str):
|
||||
self.ccode = self.function_deco + proc_name + self.function_head
|
||||
|
||||
def postproc_end(self, proc_name: str):
|
||||
self.procs.append(self.ccode + 'return 0;\n}')
|
||||
self.ccode = ''
|
||||
self.queries.append('P' + proc_name)
|
||||
|
||||
def finalize_udf(self):
|
||||
if self.udf is not None:
|
||||
return (Context.udf_head
|
||||
+ self.module_stubs
|
||||
+ self.get_init_func()
|
||||
+ self.udf
|
||||
)
|
||||
else:
|
||||
return None
|
||||
|
||||
def finalize(self):
|
||||
from aquery_config import build_driver, os_platform
|
||||
if not self.finalized:
|
||||
headers = ''
|
||||
if build_driver == 'MSBuild':
|
||||
headers ='#include \"./server/pch.hpp\"\n'
|
||||
|
||||
for h in self.headers:
|
||||
if h[0] != '"':
|
||||
headers += '#include <' + h + '>\n'
|
||||
else:
|
||||
headers += '#include ' + h + '\n'
|
||||
if os_platform == 'win':
|
||||
headers += '#undef max\n'
|
||||
headers += '#undef min\n'
|
||||
|
||||
self.ccode = headers + '\n'.join(self.procs)
|
||||
self.headers = set()
|
||||
return self.ccode
|
||||
|
||||
Reference in New Issue
Block a user