udf support for monetdbe backend

reconstruct/__init__.py

@@ -1,11 +1,31 @@
 from reconstruct.ast import Context, ast_node
+saved_cxt = None
 
-def initialize():
-    return Context()    
+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)
-            
-__all__ = ["initialize", "generate"]
+
+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)
+    cxt.print(cxt.sql)
+    return cxt
+    
+__all__ = ["initialize", "generate", "exec", "saved_cxt"]

reconstruct/ast.py

@@ -1,4 +1,7 @@
-from engine.utils import enlist, base62uuid, base62alp
+from re import T
+from typing import Set, Tuple
+from engine.types import *
+from engine.utils import enlist, base62uuid, base62alp, get_leagl_name
 from reconstruct.storage import Context, TableInfo, ColRef
     
 class ast_node:
@@ -76,30 +79,60 @@ class projection(ast_node):
         # deal with projections
         self.out_table = TableInfo('out_'+base62uuid(4), [], self.context)
         cols = []
-        col_exprs = []
+        col_ext : Set[ColRef]= set()
+        col_exprs : List[Tuple[str, Types]] = []
+        
+        proj_map = dict()
+        var_table = dict()
+        
         for i, proj in enumerate(self.projections):
             compound = False
             self.datasource.rec = set()
             name = ''
+            this_type = AnyT
             if type(proj) is dict:
-                
                 if 'value' in proj:
                     e = proj['value']
-                    name = expr(self, e).sql
-                    disp_name = ''.join([a if a in base62alp else '' for a in name])
+                    proj_expr = expr(self, e)
+                    this_type = proj_expr.type
+                    name = proj_expr.sql
                     compound = True # compound column
+                    if not proj_expr.is_special:
+                        y = lambda x:x
+                        name = eval('f\'' + name + '\'')
+                        if name not in var_table:
+                            var_table[name] = len(col_exprs)
+                        proj_map[i] = [this_type, len(col_exprs)]
+                        col_exprs.append((name, proj_expr.type))
+                    else:
+                        self.context.headers.add('"./server/aggregations.h"')
+                        if self.datasource.rec is not None:
+                            col_ext = col_ext.union(self.datasource.rec)
+                        proj_map[i] = [this_type, proj_expr.sql]
+                        
                     if 'name' in proj: # renaming column by AS keyword
-                        name += ' ' +  proj['name']
-                    col_exprs.append(name)
-                
+                        name += ' AS ' +  proj['name']
+                        if not proj_expr.is_special:
+                            var_table[proj['name']] = len(col_exprs)
+                    
+                    disp_name = get_leagl_name(name)
+                    
             elif type(proj) is str:
                 col = self.datasource.get_col(proj)
+                this_type = col.type
                 name = col.name
             self.datasource.rec = None
             # TODO: Type deduction in Python
-            cols.append(ColRef('unknown', self.out_table, None, disp_name, i, compound=compound))
-        self.add(', '.join(col_exprs))
-        
+            cols.append(ColRef(this_type, self.out_table, None, disp_name, i, compound=compound))
+        col_ext = [c for c in col_ext if c.name not in var_table] # remove duplicates in var_table
+        col_ext_names = [c.name for c in col_ext]
+        self.add(', '.join([c[0] for c in col_exprs] + col_ext_names))
+    
+        _base_offset = len(col_exprs)
+        for i, col in enumerate(col_ext_names):
+            if col not in var_table:
+                var_table[col] = i + _base_offset
+    
         def finialize(astnode:ast_node):
             if(astnode is not None):
                 self.add(astnode.sql)
@@ -110,13 +143,50 @@ class projection(ast_node):
         if 'orderby' in node:
             self.add(orderby(self, node['orderby']).sql)
         if 'outfile' in node:
-            self.add(outfile(self, node['outfile']).sql)
+            self.sql = outfile(self, node['outfile'], sql = self.sql).sql
         if self.parent is None:
             self.emit(self.sql+';\n')
         else: 
             # TODO: subquery, name create tmp-table from subquery w/ alias as name 
             pass
+        # cpp module codegen
+        self.context.has_dll = True
+        # extract typed-columns from result-set
+        vid2cname = [0]*len(var_table)
+        pyname2cname = dict()
+        typenames = [c[1] for c in col_exprs] + [c.type for c in col_ext]
+        length_name = 'len_' + base62uuid(6)
+        self.context.emitc(f'auto {length_name} = server->cnt;')
         
+        for v, idx in var_table.items():
+            vname = get_leagl_name(v) + '_' + base62uuid(3)
+            pyname2cname[v] = vname
+            self.context.emitc(f'auto {vname} = ColRef<{typenames[idx].cname}>({length_name}, server->getCol({idx}));')
+            vid2cname[idx] = vname
+        # Create table into context
+        outtable_name = 'out_' + base62uuid(6)
+        out_typenames = [None] * len(proj_map)
+        for key, val in proj_map.items():
+            if type(val[1]) is str:
+                x = True
+                y = lambda t: pyname2cname[t]
+                val[1] = eval('f\'' + val[1] + '\'')
+            if val[0] == LazyT:
+                out_typenames[key] = f'value_type<decays<decltype({val[1]})>>'
+            else:
+                out_typenames[key] = val[0].cname
+            
+        # out_typenames = [v[0].cname for v in proj_map.values()]
+        self.context.emitc(f'auto {outtable_name} = new TableInfo<{",".join(out_typenames)}>("{outtable_name}");')
+        
+        for key, val in proj_map.items():
+            if type(val[1]) is int:
+                self.context.emitc(f'{outtable_name}->get_col<{key}>().initfrom({vid2cname[val[1]]});')
+            else:
+        # for funcs evaluate f_i(x, ...)
+                self.context.emitc(f'{outtable_name}->get_col<{key}>() = {val[1]};')
+        # print out col_is
+        self.context.emitc(f'print(*{outtable_name});')
         
 class orderby(ast_node):
     name = 'order by'
@@ -147,6 +217,7 @@ class join(ast_node):
         self.joins:list = []
         self.tables = []
         self.tables_dir = dict()
+        self.rec = None
         # self.tmp_name = 'join_' + base62uuid(4)
         # self.datasource = TableInfo(self.tmp_name, [], self.context)
     def append(self, tbls, __alias = ''):
@@ -169,7 +240,7 @@ class join(ast_node):
     def produce(self, node):
         if type(node) is list:
             for d in node:
-                self.append(join(self, d).__str__())
+                self.append(join(self, d))
         
         elif type(node) is dict:
             alias = ''
@@ -205,7 +276,10 @@ class join(ast_node):
     def get_cols(self, colExpr: str) -> ColRef:
         for t in self.tables:
             if colExpr in t.columns_byname:
-                return t.columns_byname[colExpr]                
+                col = t.columns_byname[colExpr]
+                if type(self.rec) is set:
+                    self.rec.add(col)
+                return col
             
     def parse_col_names(self, colExpr:str) -> ColRef:
         parsedColExpr = colExpr.split('.')
@@ -245,7 +319,7 @@ class create_table(ast_node):
         self.sql = f'CREATE TABLE {tbl.table_name}('
         columns = []
         for c in tbl.columns:
-            columns.append(f'{c.name} {c.type.upper()}')
+            columns.append(f'{c.name} {c.type.sqlname}')
         self.sql += ', '.join(columns)
         self.sql += ')'
         if self.context.use_columnstore:
@@ -275,7 +349,13 @@ class insert(ast_node):
 class load(ast_node):
     name="load"
     first_order = name
-    def produce(self, node):
+    def init(self, _):
+        if self.context.dialect == 'MonetDB':
+            self.produce = self.produce_monetdb
+        else:
+            self.produce = self.produce_aq
+            
+    def produce_aq(self, node):
         node = node['load']
         s1 = 'LOAD DATA INFILE '
         s2 = 'INTO TABLE '
@@ -284,16 +364,97 @@ class load(ast_node):
         if 'term' in node:
             self.sql += f' {s3} \"{node["term"]["literal"]}\"'
             
-            
+    def produce_monetdb(self, node):
+        node = node['load']
+        s1 = f'COPY OFFSET 2 INTO {node["table"]} FROM '
+        s2 = ' ON SERVER '
+        s3 = ' USING DELIMITERS '
+        import os
+        p = os.path.abspath(node['file']['literal']).replace('\\', '/')
+        self.sql = f'{s1} \'{p}\' {s2} '
+        if 'term' in node:
+            self.sql += f' {s3} \'{node["term"]["literal"]}\''
+                    
 class outfile(ast_node):
     name="_outfile"
-    def produce(self, node):
+    def __init__(self, parent, node, context = None, *, sql = None):
+        super().__init__(parent, node, context)
+        self.sql = sql
+        if self.context.dialect == 'MonetDB':
+            self.produce = self.produce_monetdb
+        else:
+            self.produce = self.produce_aq
+            
+    def produce_aq(self, node):
         filename = node['loc']['literal'] if 'loc' in node else node['literal']
-        self.sql = f'INTO OUTFILE "{filename}"'
+        self.sql += f'INTO OUTFILE "{filename}"'
         if 'term' in node:
             self.sql += f' FIELDS TERMINATED BY \"{node["term"]["literal"]}\"'
 
+    def produce_monetdb(self, node):
+        filename = node['loc']['literal'] if 'loc' in node else node['literal']
+        import os
+        p = os.path.abspath('.').replace('\\', '/') + '/' + filename
+        self.sql = f'COPY {self.sql} INTO "{p}"'
+        d = '\t'
+        e = '\n'
+        if 'term' in node:
+            d = node['term']['literal']
+        self.sql += f' delimiters \'{d}\', \'{e}\''
 
+class udf(ast_node):
+    name = 'udf'
+    first_order = name
+    def __call__(self, c_code = False, *args):
+        from engine.types import fn_behavior
+        return fn_behavior(self, c_code, *args)
+    def return_type(self, *_ : Types):
+        return LazyT
+    def init(self, node):
+        self.var_table = {}
+        self.args = []
+        if self.context.udf is None:
+            self.context.udf = Context.udf_head
+            self.context.headers.add('\"./udf.hpp\"')
+            
+    def produce(self, node):
+        from engine.utils import get_leagl_name, check_leagl_name
+        node = node[self.name]
+        # register udf
+        self.cname = get_leagl_name(node['fname'])
+        self.sqlname = self.cname
+        self.context.udf_map[self.cname] = self
+        self.ccode = f'auto {self.cname} = []('
+        
+    def consume(self, node):
+        from engine.utils import get_leagl_name, check_leagl_name
+        node = node[self.name]
+        
+        if 'params' in node:
+            for args in node['params']:
+                cname = get_leagl_name(args)
+                self.var_table[args] = cname
+                self.args.append(cname)
+        self.ccode += ', '.join([f'auto {a}' for a in self.args]) + ') {\n'
+        
+        if 'assignment' in node:
+            for assign in node['assignment']:
+                var = assign['var']
+                ex = expr(self, assign['expr'], c_code=True)
+                if var in self.var_table:
+                    self.ccode += f'\t{var} = {ex.eval()};\n'
+                else:
+                    cvar = get_leagl_name(var)
+                    self.var_table[var] = cvar
+                    self.ccode += f'\tauto {cvar} = {ex.eval()};\n'
+                    
+        ret = node['ret']
+        self.ccode += f'\treturn {expr(self, ret, c_code=True).eval()};'
+        self.ccode += '\n};\n'
+        print(self.ccode)
+        
+        self.context.udf += self.ccode + '\n'
+    
 def include(objs):
     import inspect
     for _, cls in inspect.getmembers(objs):

reconstruct/expr.py

@@ -1,56 +1,33 @@
 from reconstruct.ast import ast_node
-from reconstruct.storage import ColRef, TableInfo
-
+from reconstruct.storage import ColRef
+from engine.types import *
 
 class expr(ast_node):
     name='expr'
-    builtin_func_maps = {
-        'max': 'MAX',
-        'min': 'MIN', 
-        'avg': 'AVG',
-        'sum': 'SUM',
-        'count' : 'COUNT',
-        'mins': ['mins', 'minw'],
-        'maxs': ['maxs', 'maxw'],
-        'avgs': ['avgs', 'avgw'],
-        'sums': ['sums', 'sumw'],
-    }
-    
-    binary_ops = {
-        'sub':'-',  
-        'add':'+', 
-        'mul':'*', 
-        'div':'/',
-        'mod':'%',
-        'and':' AND ',
-        'or':' OR ',
-        'xor' : ' XOR ',
-        'gt':'>',
-        'lt':'<',
-        'le':'<=',
-        'gt':'>='
-    }
 
-    compound_ops = {
-    }
-
-    unary_ops = {
-        'neg' : '-',
-        'not' : ' NOT '
-    }
-    
-    coumpound_generating_ops = ['avgs', 'mins', 'maxs', 'sums'] + \
-       list(binary_ops.keys()) + list(compound_ops.keys()) + list(unary_ops.keys() )
-
-    def __init__(self, parent, node):
+    def __init__(self, parent, node, *, c_code = None):
+        from reconstruct.ast import projection
+        
+        self.type = None
         self.raw_col = None
         self.inside_agg = False
+        self.is_special = False
         if(type(parent) is expr):
             self.inside_agg = parent.inside_agg
+        if(type(parent) is not expr):
+            self.root = self
+            self.c_code = type(parent) is projection
+        else:
+            self.root = parent.root
+            self.c_code = parent.c_code
+        
+        if type(c_code) is bool:
+            self.c_code = c_code
+        
         ast_node.__init__(self, parent, node, None)
 
     def init(self, _):
-        from engine.projection import projection
+        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
@@ -59,70 +36,73 @@ class expr(ast_node):
         else:
             self.datasource = self.context.datasource
         self.udf_map = parent.context.udf_map
-        self.func_maps = {**self.udf_map, **self.builtin_func_maps}
-
+        self.func_maps = {**builtin_func, **self.udf_map}
+        self.operators = {**builtin_operators, **self.udf_map}
     def produce(self, node):
+        from engine.utils import enlist
         if type(node) is dict:
             for key, val in node.items():
-                if key in self.func_maps:
-                    # TODO: distinguish between UDF agg functions and other UDF functions.
-                    self.inside_agg = True
-                    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.sql += f"{cfunc}(" 
-                        for i, p in enumerate(val):
-                            self.sql += expr(self, p).sql + (',' if i < len(val) - 1 else '')
-                    else:
-                        funcname = self.func_maps[key]
-                        funcname = funcname[0] if type(funcname) is list else funcname
-                        self.sql += f"{funcname}(" 
-                        self.sql += expr(self, val).sql
-                    self.sql += ')'
-                    self.inside_agg = False
-                elif key in self.binary_ops:
-                    l = expr(self, val[0]).sql
-                    r = expr(self, val[1]).sql
-                    self.sql += 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).sql)
-                    self.sql = self.compound_ops[key][1](x)
-                elif key in self.unary_ops:
-                    self.sql += f'{self.unary_ops[key]}({expr(self, val).sql})'
+                if key in self.operators:
+                    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]
+                    
+                    self.type = op.return_type(*type_vals)
+                    self.sql = op(self.c_code, *str_vals)
+                    special_func = [*self.context.udf_map.keys(), "maxs", "mins", "avgs", "sums"]
+                    if key in special_func and not self.is_special:
+                        self.is_special = True
+                        p = self.parent
+                        while type(p) is expr and not p.is_special:
+                            p.is_special = True
+                            p = p.parent
                 else:
                     print(f'Undefined expr: {key}{val}')
 
-                if key in self.coumpound_generating_ops and not self.is_compound:
-                    self.is_compound = True
-                    p = self.parent
-                    while type(p) is expr and not p.is_compound:
-                        p.is_compound = True
-                        p = p.parent
-
         elif type(node) is str:
             p = self.parent
             while type(p) is expr and not p.isvector:
                 p.isvector = True
                 p = p.parent
-            
-            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.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.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.sql = '1' if node else '0'
+            self.type = IntT
+            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:
+                self.type = LongT
+            elif type(node) is float:
+                self.type = DoubleT
+
+    def finalize(self, y):
+        if self.c_code:
+            x = self.is_special
+            self.sql = eval('f\'' + self.sql + '\'')
+        
     def __str__(self):
         return self.sql
     def __repr__(self):
         return self.__str__()
-    
-    
+    def eval(self):
+        x = self.c_code
+        y = lambda x: x
+        return eval('f\'' + self.sql + '\'')    
+
+import engine.expr as cexpr

reconstruct/storage.py

@@ -1,6 +1,12 @@
+from engine.types import *
+
 class ColRef:
     def __init__(self, _ty, cobj, table:'TableInfo', name, id, compound = False):
-        self.type = _ty
+        self.type : Types = AnyT
+        if type(_ty) is str:
+            self.type = builtin_types[_ty.lower()]
+        elif type(_ty) is Types:
+            self.type = _ty
         self.cobj = cobj
         self.table = table
         self.name = name
@@ -19,7 +25,6 @@ class ColRef:
     def __setitem__(self, key, value):
         self.__arr__[key] = value
 
-
 class TableInfo:
     def __init__(self, table_name, cols, cxt:'Context'):
         # statics
@@ -39,6 +44,7 @@ class TableInfo:
     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']
         if new:
@@ -56,10 +62,14 @@ class TableInfo:
             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:
-            return self.columns_byname[colExpr]
+            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:
@@ -69,8 +79,15 @@ class TableInfo:
             
             
 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
+              
     def __init__(self):
-        self.sql = ''
         self.tables_byname = dict()
         self.col_byname = dict()
         self.tables = []
@@ -78,13 +95,41 @@ class Context:
         self.datasource = None
         self.udf_map = dict()
         self.use_columnstore = False
+        self.print = print
+        self.has_dll = False
+        self.dialect = 'MonetDB'
+        
+        self.new()
+
         
     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
     
-    
+    function_head = '''
+    extern "C" int __DLLEXPORT__ dllmain(Context* cxt) { 
+        using namespace std;
+        using namespace types;
+        auto server = static_cast<Server*>(cxt->alt_server);
+    '''
+    udf_head = ('#pragma once\n'
+        '#include \"./server/libaquery.h\"\n'
+        '#include \"./server/aggregations.h\"\n\n'
+        )
+    
+    def finalize(self):
+        if not self.finalized:
+            headers = ''
+            for h in self.headers:
+                if h[0] != '"':
+                    headers += '#include <' + h + '>\n'
+                else:
+                    headers += '#include ' + h + '\n'
+            self.ccode = headers + self.function_head + self.ccode + 'return 0;\n}'
+            self.headers = set()
+        return self.ccode