impl possible rec, non shared,only select cte

lib/prelude.ml

@@ -9,6 +9,11 @@ let flip f x y = f y x
 let tuck l x = l := x :: !l
 let option_list = function Some x -> [x] | None -> []
 
+let get_lazy_opt default_fn x =
+  match x with
+  | None -> default_fn ()
+  | Some y -> y
+
 let fail fmt = Printf.ksprintf failwith fmt
 let failed ~at fmt = Printf.ksprintf (fun s -> raise (At (at, Failure s))) fmt
 let printfn fmt = Printf.ksprintf print_endline fmt

lib/sql.ml

@@ -374,6 +374,10 @@ type select_result = (schema * param list)
 
 type direction = [ `Fixed | `Param of param_id ] [@@deriving show]
 
+type cte_supported_compound_op = [ `Union | `Union_all ] [@@deriving show]
+
+type compound_op = [ cte_supported_compound_op | `Except | `Intersect ] [@@deriving show]
+
 type int_or_param = [`Const of int | `Limit of param]
 type limit_t = [ `Limit | `Offset ]
 type col_name = {
@@ -392,7 +396,7 @@ and select = {
   having : expr option;
 }
 and select_full = {
-  select : select * select list;
+  select : select * (compound_op * select) list;
   order : order;
   limit : limit option;
 }
@@ -431,6 +435,8 @@ type insert_action =
   on_duplicate : assignments option;
 }
 
+type cte = { cte_name: string;  cols: string list option; stmt: select_full; }
+
 type stmt =
 | Create of table_name * [ `Schema of schema | `Select of select_full ]
 | Drop of table_name
@@ -445,6 +451,7 @@ type stmt =
 | UpdateMulti of nested list * assignments * expr option
 | Select of select_full
 | CreateRoutine of table_name * Type.kind option * (string * Type.kind * expr option) list (* table_name represents possibly namespaced function name *)
+| Cte_select of { ctes: cte list; stmt: select_full; is_recursive: bool; }
 
 (*
 open Schema

lib/sql_lexer.mll

@@ -193,6 +193,7 @@ let keywords =
    "inplace", INPLACE;
    "algorithm", ALGORITHM;
    "copy", COPY;
+   "recursive", RECURSIVE;
   ] in (* more *)
   let all token l = k := !k @ List.map (fun x -> x,token) l in
   all DATETIME_FUNC ["current_date";"current_timestamp";"current_time";"localtime";"localtimestamp";"now";];

lib/sql_parser.mly

@@ -42,7 +42,7 @@
        CASE WHEN THEN ELSE END CHANGE MODIFY DELAYED ENUM FOR SHARE MODE LOCK
        OF WITH NOWAIT ACTION NO IS INTERVAL SUBSTRING DIV MOD CONVERT LAG LEAD OVER
        FIRST_VALUE LAST_VALUE NTH_VALUE PARTITION ROWS RANGE UNBOUNDED PRECEDING FOLLOWING CURRENT ROW
-       CAST GENERATED ALWAYS VIRTUAL STORED STATEMENT DOUBLECOLON INSTANT INPLACE COPY ALGORITHM
+       CAST GENERATED ALWAYS VIRTUAL STORED STATEMENT DOUBLECOLON INSTANT INPLACE COPY ALGORITHM RECURSIVE
 %token FUNCTION PROCEDURE LANGUAGE RETURNS OUT INOUT BEGIN COMMENT
 %token MICROSECOND SECOND MINUTE HOUR DAY WEEK MONTH QUARTER YEAR
        SECOND_MICROSECOND MINUTE_MICROSECOND MINUTE_SECOND
@@ -92,6 +92,8 @@ if_exists: IF EXISTS {}
 temporary: either(GLOBAL,LOCAL)? TEMPORARY { }
 assign: name=IDENT EQUAL e=expr { name, e }
 
+cte: cte_name=IDENT cols=maybe_parenth(sequence(IDENT))? AS LPAREN stmt=select_stmt RPAREN { { cte_name; cols;stmt } }
+
 statement: CREATE ioption(temporary) TABLE ioption(if_not_exists) name=table_name schema=table_definition
               {
                 Create (name,`Schema schema)
@@ -170,6 +172,7 @@ statement: CREATE ioption(temporary) TABLE ioption(if_not_exists) name=table_nam
                 Function.add (List.length params) (Ret Any) name.tn; (* FIXME void *)
                 CreateRoutine (name, None, params)
               }
+         | is_recursive=cte_with ctes=commas(cte) stmt=select_stmt { Cte_select { ctes; stmt; is_recursive }}     
 
 parameter_default_: DEFAULT | EQUAL { }
 parameter_default: parameter_default_ e=expr { e }
@@ -200,7 +203,11 @@ parser_state_ignore: { Parser_state.mode_ignore () }
 parser_state_normal: { Parser_state.mode_normal () }
 parser_state_ident: { Parser_state.mode_ident () }
 
-select_stmt: select_core other=list(preceded(compound_op,select_core)) o=loption(order) lim=limit_t? select_row_locking?
+cte_with:
+  WITH { false }
+  | WITH RECURSIVE { true }
+
+select_stmt: select_core other=list(pair(compound_op,select_core)) o=loption(order) lim=limit_t? select_row_locking?
               {
                 { select = ($1, other); order=o; limit=lim; }
               }
@@ -538,7 +545,11 @@ sql_type: t=sql_type_flavor
         | t=sql_type_flavor LPAREN INTEGER COMMA INTEGER RPAREN
         { t }
 
-compound_op: UNION ALL? | EXCEPT | INTERSECT { }
+compound_op: 
+  | UNION { `Union }
+  | UNION ALL { `Union_all }
+  | EXCEPT { `Except } 
+  | INTERSECT { `Intersect }
 
 strict_type:
     | T_TEXT     { Text }

lib/syntax.ml

@@ -10,12 +10,21 @@ let debug = ref false
 type env = {
   tables : Tables.table list;
   schema : table_name Schema.Source.t;
+  ctes : Tables.table list;
   insert_schema : Schema.t;
   (* it is used to apply non-null comparison semantics inside WHERE expressions *)
   set_tyvar_strict: bool;
   query_has_grouping: bool;
 }
 
+module Tables_with_derived = struct
+  open Tables
+
+  let get ~env name = get_from  (env.ctes @ Tables.all()) name
+
+  let get_from ~env name = get_from  (env.ctes @ env.tables) name
+end
+
 (* expr with all name references resolved to values or "functions" *)
 type res_expr =
   | ResValue of Type.t (** literal value *)
@@ -34,12 +43,13 @@ let empty_env = { query_has_grouping = false;
   tables = []; schema = []; 
   insert_schema = []; 
   set_tyvar_strict = false; 
+  ctes = [];
 }
 
 let flat_map f l = List.flatten (List.map f l)
 
-let schema_of tables name =
-  let result = Tables.get_from tables name in 
+let schema_of ~env name =
+  let result = Tables_with_derived.get_from ~env name in 
   List.map (fun attr -> { Schema.Source.Attr.sources=[result |> fst]; attr; }) (result |> snd)
 
 let get_or_failwith = function `Error s -> failwith s | `Ok t -> t
@@ -114,7 +124,7 @@ let all_columns  = make_unique $ Schema.cross_all
 
 let all_tbl_columns = all_columns $ List.map snd
 
-let resolve_column tables schema {cname;tname} =
+let resolve_column ~env {cname;tname} =
   let open Schema.Source in
   let open Attr in
   let by_name_and_sources tname name source_attr = source_attr.attr.name = 
@@ -126,23 +136,23 @@ let resolve_column tables schema {cname;tname} =
     | [x] -> Some x
     | [] -> None
     | list -> Some (List.last list) in
-  let result = find schema cname in
+  let result = find env.schema cname in
   let find_by_name t name = List.find_all (by_name name) t in
   let find t name =
     let err_data = from_schema t in
     match find_by_name t name with
     | [x] -> x
     | [] -> raise (Schema.Error (err_data,"missing attribute : " ^ name))
     | _ -> raise (Schema.Error (err_data,"duplicate attribute : " ^ name)) in
-  let default_result = find (Option.map_default (schema_of tables) schema tname) cname in  
-  Option.default default_result result
+  let default_result () = find (Option.map_default (schema_of ~env) env.schema tname) cname in  
+  get_lazy_opt default_result result
 
-let resolve_column_assignments tables l =
+let resolve_column_assignments ~env l =
   let open Schema.Source in 
   let open Attr in
-  let all = all_tbl_columns (List.map (fun (a, b) -> a, (List.map (fun attr -> {sources=[a]; attr}) b)) tables) in
+  let all = all_tbl_columns (List.map (fun (a, b) -> a, (List.map (fun attr -> {sources=[a]; attr}) b)) env.tables) in
   l |> List.map begin fun (col,expr) ->
-    let attr = resolve_column tables all col in
+    let attr = resolve_column ~env:{ env with schema = all } col in
     (* autoincrement is special - nullable on insert, strict otherwise *)
     let typ = if Constraints.mem Autoincrement attr.attr.extra then 
       Sql.Type.nullable attr.attr.domain.t else attr.attr.domain in
@@ -154,11 +164,11 @@ let resolve_column_assignments tables l =
     | _ -> equality typ expr
   end
 
-let get_columns_schema tables l =
-  let all = all_tbl_columns (List.map (fun (a, b) -> a, (List.map (fun attr -> {Schema.Source.Attr.sources=[a]; attr}) b)) tables) in
+let get_columns_schema ~env l =
+  let all = all_tbl_columns (List.map (fun (a, b) -> a, (List.map (fun attr -> {Schema.Source.Attr.sources=[a]; attr}) b)) env.tables) in
   (* FIXME col_name *)
   l |> List.map (fun col -> 
-    let res = resolve_column tables all col in
+    let res = resolve_column ~env:{ env with schema = all } col in
     { res with attr = { res.attr with name = col.cname } })
 
 let _print_env env =
@@ -191,7 +201,7 @@ let rec resolve_columns env expr =
   let rec each e =
     match e with
     | Value x -> ResValue x
-    | Column col -> ResValue (resolve_column env.tables env.schema col).attr.domain
+    | Column col -> ResValue (resolve_column ~env col).attr.domain
     | Inserted name ->
       let attr = try Schema.find env.insert_schema name with Schema.Error (_,s) -> fail "for inserted values : %s" s in
       ResValue attr.domain
@@ -376,11 +386,11 @@ and infer_schema env columns =
 (*   let all = tables |> List.map snd |> List.flatten in *)
   let resolve1 = function
     | All -> env.schema
-    | AllOf t -> schema_of env.tables t
+    | AllOf t -> schema_of ~env t
     | Expr (e,name) ->
       let col =
         match e with
-        | Column col -> resolve_column env.tables env.schema col
+        | Column col -> resolve_column ~env col
         | _ -> { attr = unnamed_attribute (resolve_types env e |> snd |> get_or_failwith); sources = [] }
       in
       let col = Option.map_default (fun n -> {col with attr = { col.attr with name = n }}) col name in
@@ -425,7 +435,7 @@ and join env ((schema,p0,ts0),joins) =
   List.fold_left do_join (env, p0) joins
 
 and params_of_assigns env ss =
-  let exprs = resolve_column_assignments env.tables ss in
+  let exprs = resolve_column_assignments ~env ss in
   get_params_l env exprs
 
 and params_of_order order final_schema tables =
@@ -504,37 +514,38 @@ and resolve_source env (x,alias) =
     if alias <> None then failwith "No alias allowed on nested tables";
     s, p, env.tables
   | `Table s ->
-    let (name,s) = Tables.get s in
+    let (name,s) = Tables_with_derived.get ~env s in
     let sources = (name :: option_list alias) in
     let s3 = List.map (fun attr -> { Schema.Source.Attr.attr; sources }) s  in
     s3, [], List.map (fun name -> name, s) sources
 
-and eval_select_full env { select=(select,other); order; limit; } =
-  let (s1,p1,tbls,cardinality) = eval_select env select in
+and eval_select_full env stmt =
+  let (s1,p1,tbls,cardinality) = eval_select env (fst @@ stmt.select) in
+  eval_compound ~env:{ env with tables = tbls; } (p1, s1, cardinality, stmt)
+
+and eval_compound ~env result = 
+  let (p1, s1, cardinality, stmt) = result in
+  let { select=(_select, other); order; limit; } = stmt in
+  let other = List.map snd other in
   let (s2l,p2l) = List.split (List.map (fun (s,p,_,_) -> s,p) @@ List.map (eval_select env) other) in
-  if false then
-    eprintf "cardinality=%s other=%u\n%!"
-            (Stmt.cardinality_to_string cardinality)
-            (List.length other);
   let cardinality = if other = [] then cardinality else `Nat in
   (* ignoring tables in compound statements - they cannot be used in ORDER BY *)
   let final_schema = List.fold_left Schema.compound s1 s2l in
-  let p3 = params_of_order order final_schema tbls in
+  let p3 = params_of_order order final_schema env.tables in
   let (p4,limit1) = match limit with Some (p,x) -> List.map (fun p -> Single p) p, x | None -> [],false in
   (*                 Schema.check_unique schema; *)
   let cardinality =
     if limit1 && cardinality = `Nat then `Zero_one
-                                    else cardinality in
-  final_schema,(p1@(List.flatten p2l)@p3@p4 : var list), Stmt.Select cardinality
-
+      else cardinality in
+  final_schema,(p1@(List.flatten p2l)@p3@p4 : var list), Stmt.Select cardinality  
 
-let update_tables sources ss w =
+let update_tables ~env sources ss w =
   let schema = Schema.cross_all @@ List.map (fun (s,_,_) -> s) sources in
   let p0 = List.flatten @@ List.map (fun (_,p,_) -> p) sources in
   let tables = List.flatten @@ List.map (fun (_,_,ts) -> ts) sources in (* TODO assert equal duplicates if not unique *)
-  let result = get_columns_schema tables (List.map fst ss) in
+  let result = get_columns_schema ~env:{ env with tables } (List.map fst ss) in
   let env = { empty_env with 
-    tables; schema; insert_schema=List.map (fun i -> i.Schema.Source.Attr.attr) result; } in
+    tables; schema; insert_schema=Schema.Source.from_schema result; } in
   let p1 = params_of_assigns env ss in
   let p2 = get_params_opt { env with set_tyvar_strict = true } w in
   p0 @ p1 @ p2
@@ -552,14 +563,15 @@ let annotate_select select types =
 
 let rec eval (stmt:Sql.stmt) =
   let open Stmt in
-  let open Schema.Source.Attr in
+  let open Schema.Source in
+  let open Attr in
   match stmt with
   | Create (name,`Schema schema) ->
       Tables.add (name, schema);
       ([],[],Create name)
   | Create (name,`Select select) ->
       let (schema,params,_) = eval_select_full empty_env select in
-      Tables.add (name, List.map (fun i -> i.attr) schema);
+      Tables.add (name, from_schema schema);
       ([],params,Create name)
   | Alter (name,actions) ->
       List.iter (function
@@ -663,18 +675,54 @@ let rec eval (stmt:Sql.stmt) =
 
     let r = List.map (fun attr -> {Schema.Source.Attr.attr; sources=[f] }) s in
 
-    let params = update_tables [r,[],[(f, s)]] ss w in
+    let params = update_tables ~env:empty_env [r,[],[(f, s)]] ss w in
     let p3 = params_of_order o [] [(f, s)] in
     [], params @ p3 @ (List.map (fun p -> Single p) lim), Update (Some table)
   | UpdateMulti (tables,ss,w) ->
     let sources = List.map (fun src -> resolve_source empty_env ((`Nested src), None)) tables in
-    let params = update_tables sources ss w in
+    let params = update_tables ~env:empty_env sources ss w in
     [], params, Update None
   | Select select -> 
     let (schema, a, b) = eval_select_full empty_env select in
-    List.map (fun i -> i.Schema.Source.Attr.attr) schema , a ,b
+    from_schema schema , a ,b
   | CreateRoutine (name,_,_) ->
     [], [], CreateRoutine name
+  | Cte_select { ctes; stmt; is_recursive } ->
+    let ctes, p1 =
+    List.fold_left
+      (fun (acc_ctes, acc_vars) cte ->
+        let env = { empty_env with ctes = acc_ctes } in
+        let tbl_name = make_table_name cte.cte_name in
+        let s1, p1, _kind =
+          if is_recursive then (
+            let { select = select, other; _ } = cte.stmt in
+            let other = List.map (fun cmb -> match fst cmb with 
+              | #cte_supported_compound_op -> cmb
+              | `Except | `Intersect -> 
+                fail "%s: Recursive table reference in EXCEPT or INTERSECT operand is not allowed in CTEs" cte.cte_name
+            ) other in
+            let stmt = { cte.stmt with select = select, other } in
+            let s1, p1, tbls, cardinality = eval_select env select in
+            let a1 = from_schema s1 in
+            eval_compound
+              ~env:{ env with tables = tbls; ctes = (tbl_name, a1) :: env.ctes }
+              (p1, s1, cardinality, stmt))
+          else eval_select_full env cte.stmt
+        in
+        let a1 = from_schema s1 in
+        let select_all = Option.default (List.map (fun i -> i.name) a1) cte.cols in
+        let s2 = 
+          try 
+            List.map2 (fun col cut -> { cut with name = col }) select_all a1 
+          with 
+            List.Different_list_size _ -> 
+              fail "%s: SELECT list and column names list have different column counts" cte.cte_name in
+        (tbl_name, s2) :: acc_ctes, p1 @ acc_vars)
+      ([], []) ctes
+    in
+    let s1, p2, b = eval_select_full { empty_env with ctes } stmt in
+    from_schema s1, p1 @ p2, b
+
 
 (* FIXME unify each choice separately *)
 let unify_params l =