Source file syntax.ml

(** SQL syntax and RA *)

open Printf
open ExtLib
open Prelude
open Sql

module Config = struct 
  let debug = ref false
  (* If strict mode is not enabled, some dbs allow this. *)
  let allow_write_notnull_null = ref false
  let dynamic_select = ref false
end

type query_scope =
  | Top_level
  | Subquery
  | From_passthrough

type env = {
  tables : Tables.table list;
  schema : table_name Schema.Source.t;
  (* 
    1. CTEs = tables for the current statement (not keeping during whole .sql)
    2. It merges with global tables during source resolving
    3. The Tables field mostly stores aliases and forms a scheme
  *)
  ctes : Tables.table list;
  (* it is used to apply non-null comparison semantics inside WHERE expressions *)
  set_tyvar_strict: bool;
  query_has_grouping: bool;
  is_order_by: bool;
  (* Check if the current query is an UPDATE statement *)
  is_update: bool;
  insert_resolved_types: (string, Type.t) Hashtbl.t; (* for INSERT .. VALUES *)
  scope: query_scope;
}

(* Merge global tables with ctes during resolving sources in SELECT .. FROM sources, JOIN *)
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

type enum_ctor_value_data = { ctor_name: string; pos: pos; } [@@deriving show]

(* expr with all name references resolved to values or "functions" *)
type res_expr =
  | ResValue of Type.t (** literal value *)
  | ResParam of Type.t param * Meta.t
  | ResSelect of Type.t * vars
  | ResInTupleList of { param_id: param_id; res_in_tuple_list: res_in_tuple_list; kind: in_or_not_in; pos: pos; }
  | ResInparam of Type.t param * Meta.t
  | ResChoices of param_id * res_expr choices
  | ResInChoice of param_id * in_or_not_in * res_expr
  | ResFun of res_fun (** function kind (return type and flavor), arguments *)
  | ResOptionActions of { choice_id: param_id; res_choice: res_expr; pos: (pos * pos); kind: Sql.option_actions_kind }
  | ResCase of { case: res_expr option; branches: case_branch list; else_: res_expr option }
  [@@deriving show] 

and case_branch = { when_: res_expr; then_: res_expr; } [@@deriving show]

and res_fun = { kind: Type.t func [@printer pp_func] ; parameters: res_expr list; is_over_clause: bool; } [@@deriving show]

and res_in_tuple_list =
  ResTyped of (Type.t * Meta.t) list | Res of (res_expr * Meta.t) list [@@deriving show]

let empty_env = { query_has_grouping = false; 
  tables = []; schema = []; 
  set_tyvar_strict = false; 
  ctes = [];
  is_order_by = false;
  is_update = false;
  insert_resolved_types = Hashtbl.create 16;
  scope = Top_level;
}

let flat_map f l = List.flatten (List.map f l)

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

let values_or_all table names =
  let schema = Tables.get_schema table in
  match names with
  | Some names -> 
    let req_missing =
      List.filter_map
        (fun { extra; name; _ } ->
          let open Constraints in
          if inter (of_list [Autoincrement; WithDefault; NotNull]) extra = of_list [NotNull]
            && not @@ List.mem name names then Some name
          else None
        )
        schema
    in
    begin match req_missing with 
    | [] -> ()
    | fields -> 
        fail "Fields: (%s) don't have a default value" (String.concat "," fields) end;    
    Schema.project names schema
  | None -> schema


let list_same l =
  match l with
  | [] -> None
  | x::xs -> if List.for_all (fun y -> x = y) xs then Some x else None

let rec is_grouping = function
| Choices (p,l) ->
  begin match list_same @@ List.map (fun (_,expr) -> Option.map_default is_grouping false expr) l with
  | None -> failed ~at:p.pos "inconsistent grouping in choice branches"
  | Some v -> v
  end
| Fun { kind ; parameters; _ } ->
  (* grouping function of zero or single parameter or function on grouping result *)
  (Sql.is_grouping kind && List.length parameters <= 1) || List.exists is_grouping parameters
| e -> List.exists is_grouping (sub_exprs e)

let is_windowing =
  expr_exists (function Sql.Fun { is_over_clause; _ } -> is_over_clause | _ -> false)

let exists_grouping columns =
  List.exists (function
    | { value = Expr ({ value; _ }, _); _ } -> is_grouping value
    | { value = (All | AllOf _); _ } -> false
  ) columns

let exists_windowing columns =
  List.exists (function
    | { value = Expr ({ value; _ }, _); _ } -> is_windowing value
    | { value = (All | AllOf _); _ } -> false
  ) columns  

(* all columns from tables, without duplicates *)
(* FIXME check type of duplicates *)
let make_unique =
  List.unique ~cmp:(fun a1 a2 ->
    (* Check if columns are from the same table (source)  *)
    a1.Schema.Source.Attr.sources = a2.sources
    && a1.attr.name = a2.attr.name
    (* Check if columns are named *)
    && a1.attr.name <> "")

let all_columns = make_unique $ Schema.cross_all

let all_tbl_columns = all_columns $ List.map snd

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 = 
    name && Option.map_default 
      (fun tname -> List.mem tname.tn (List.map(fun i -> i.tn) source_attr.sources)) false tname in
  let find_by t name = List.find_all (by_name_and_sources tname name) t in
  let find t name =
    match find_by t name with
    | [x] -> Some x
    | [] -> None
    | list -> Some (List.last list) 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  
  match result with 
  | None -> find (Option.map_default (schema_of ~env) env.schema tname) cname
  | Some result -> result

let resolve_column_opt ~env col =
  match resolve_column ~env col with
  | attr -> Some attr
  | exception (Schema.Error _ | Failure _) -> None

let rec merge_meta_into_params ~shallow meta expr =
  match expr with
  | Param (p, m) -> Param (p, Meta.merge_right meta m)
  | Inparam (p, m) -> Inparam (p, Meta.merge_right meta m)
  | OptionActions ({ choice; _ } as o) ->
    OptionActions { o with choice = merge_meta_into_params ~shallow meta choice }
  | e when shallow -> e
  | e -> map_sub_exprs (merge_meta_into_params ~shallow meta) e

let set_param_meta ~env col e = 
  let m' = (resolve_column ~env col).attr.meta in
  merge_meta_into_params ~shallow:true m' e

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)) env.tables) in
  let env = { env with schema = all } in
  l |> List.map begin fun (col,expr) ->
    let attr = resolve_column ~env col in
    let is_non_nullifiable = env.is_update && Sql.Meta.get_is_non_nullifiable attr.attr.meta in
    (* non nullifiable: once a column value is set to non-NULL, it can never be updated back to NULL *)
    let attr = 
      if is_non_nullifiable then
        { attr with 
          attr = { attr.attr with 
            domain = Type.make_strict attr.attr.domain;
            extra = Constraints.add NotNull attr.attr.extra 
          }
        }
      else attr 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
    let typ = Sql.make_collated ~collated:typ () in
    if !Config.debug then eprintfn "column assignment %s type %s" col.cname (Type.show typ.collated);
    (* add equality on param and column type *)
    let equality typ expr = Fun { fn_name = "col_assign"; kind = (Col_assign { ret_t = Var 0; col_t = Var 0; arg_t = Var 0 }); parameters = [Value typ; expr]; is_over_clause = false } in
    let with_default assign = if not @@ Constraints.mem WithDefault attr.attr.extra then fail "Column %s doesn't have default value" col.cname else assign in
    match expr with
    | RegularExpr (Choices (n,l)) ->
      (* Apply metadata to params inside each choice branch *)
      let l_with_meta = List.map (fun (n,e) -> n, Option.map (set_param_meta ~env col) e) l in
      Choices (n, List.map (fun (n,e) -> n, Option.map (equality typ) e) l_with_meta)
    | RegularExpr (OptionActions ch) ->
      OptionActions { ch with choice = (equality typ) ch.choice }  (* FIXME hack, should propagate properly *)
    | RegularExpr expr -> equality typ (set_param_meta ~env col expr)
    | WithDefaultParam (e, pos) -> with_default @@ OptionActions { choice = equality typ (set_param_meta ~env col e); pos; kind = SetDefault }
    | AssignDefault -> with_default @@ (Value typ)
  end

let _print_env env =
  eprintfn "env: ";
  Sql.Schema.print @@ Schema.Source.from_schema env.schema;
  Tables.print stderr env.tables

let update_schema_with_aliases all_schema final_schema = 
  let applied = all_schema |> List.filter (fun s1 -> List.for_all Schema.Source.Attr.(fun s2 -> s2.attr.name <> s1.attr.name) final_schema) in  
  applied @ final_schema

let rec bool_choice_id = function
  | Inparam (p, _)
  | Param (p, _) -> Some p.id
  | Choices (p, _)
  | InTupleList { value = { param_id = p; _ }; _ }
  | InChoice(p, _, _) -> Some p
  | OptionActions _ -> None
  | e -> List.find_map bool_choice_id (sub_exprs e)

let extract_meta_from_col ~env expr = 
  let rec aux = function 
    (* col_name = @param *)
    | Sql.Fun ({ parameters = ([Column a; b]); kind = Comparison _; _ } as fn)
    (* col_name IN @param *)
    | Fun ({ parameters = ([Column a; (Inparam _) as b]); _ } as fn) -> 
      Fun { fn with parameters = [Column a; set_param_meta ~env a.collated b] }
    | Sql.Fun ({ parameters = ([b; Column a]); kind = Comparison _; _ } as fn)
    (* col_name IN @param *)
    | Fun ({ parameters = ([(Inparam _) as b; Column a;]); _ } as fn) -> 
      Fun { fn with parameters = [set_param_meta ~env a.collated b; Column a;] }
    | e -> map_sub_exprs aux e
  in
  aux expr    

let dynamic_allowed env =
  !Config.dynamic_select &&
  match env.scope with
  | Top_level | From_passthrough -> true
  | Subquery -> false

let dynamic_col_param_name = "col"

let make_dynamic_select ~env columns =
  if not (dynamic_allowed env) then
    columns
  else
    let module S = Set.Make(String) in
    let unique_name used base =
      if not (S.mem base used) then
        base
      else
        let rec aux n =
          let candidate = base ^ "_" ^ string_of_int n in
          if S.mem candidate used then aux (n + 1) else candidate
        in
        aux 1
    in
    let use_expanded_choices ~used ~idx ~column_pos ~schema =
      let rev_choices, used, idx =
        List.fold_left (fun (choices, used, idx) { Schema.Source.Attr.attr = { name; _ }; sources } ->
          let source = match sources with s :: _ -> Some s | [] -> None in
          let col_name = unique_name used name in
          let expr = Column { collated = { cname = name; tname = source }; collation = None } in
          let choice = ({ value = Some col_name; pos = Sql.dummy_pos }, Some expr), column_pos in
          choice :: choices, S.add col_name used, idx + 1
        ) ([], used, idx) schema
      in
      (used, idx, snd column_pos), List.rev rev_choices
    in
    let (_, _, last_col_end), choices_chunks =
      List.fold_left_map (fun (used, idx, _last_end) column ->
        match column.value with
        | Expr ({ value = e; pos = ep_start, ep_end }, alias) ->
          let base_name = Option.default begin match e with
            | Column { collated = { cname; _ }; _ } -> cname
            | _ -> dynamic_col_param_name ^ string_of_int (idx + 1)
            end alias
          in
          let col_name = unique_name used base_name in
          let choice = (({ value = Some col_name; pos = (ep_start, ep_end) }, Some e), column.pos) in
          ((S.add col_name used, idx + 1, snd column.pos), [choice])
        | All ->
          use_expanded_choices ~used ~idx ~column_pos:column.pos ~schema:env.schema
        | AllOf t ->
          use_expanded_choices ~used ~idx ~column_pos:column.pos ~schema:(schema_of ~env t)
      ) (S.empty, 0, 0) columns
    in
    let all_choices = List.concat choices_chunks in
    match all_choices with
    | [] -> columns
    | (_, (first_pos, _)) :: _ ->
      let outer_pos = (first_pos, last_col_end) in
      let choices = List.map fst all_choices in
      [{ value = Expr ({ value = Choices ({ value = Some dynamic_col_param_name; pos = outer_pos }, choices); pos = outer_pos }, None); pos = outer_pos }]
  


type resolved_source = {
  rsrc_schema : table_name Schema.Source.t;
  rsrc_params : vars;
  rsrc_tables : Tables.table list;
  rsrc_dynamic : schema_column_with_sources list;
  rsrc_physical_table : Sql.join_source option;
}

module From = struct
  type join = {
    src : resolved_source;
    kind : Schema.Join.typ;
    cond : join_condition;
    pos : pos;
  }

  type t = {
    base : resolved_source;
    joins : join list;
  }

  let dynamic_columns from =
    let sources { base; joins } = base :: List.map (fun j -> j.src) joins in
    List.concat_map (fun src -> src.rsrc_dynamic) (Option.map_default sources [] from)
end

module Table_refs : sig
  type t
  val of_expr : env:env -> Sql.expr -> t
  val of_exprs : env:env -> Sql.expr list -> t
  val may_refer : Sql.join_source -> t -> bool
end = struct
  module Names = Set.Make(String)

  type t = Names.t option

  let anything = None

  let empty = Some Names.empty

  let union a b =
    match a, b with
    | Some x, Some y -> Some (Names.union x y)
    | None, _ | _, None -> anything

  let of_attr attr =
    Names.of_list (List.map (fun (s : table_name) -> s.tn) attr.Schema.Source.Attr.sources)

  let rec of_expr ~env = function
    | Sql.Column c -> Option.map of_attr (resolve_column_opt ~env c.collated)
    | SelectExpr _ -> anything
    | e -> of_exprs ~env (sub_exprs e)

  and of_exprs ~env l = List.fold_left (fun acc e -> union acc (of_expr ~env e)) empty l

  let may_refer source =
    Option.map_default (Names.mem (Sql.join_source_name source).tn) true
end

let matches_at_most_one_row ~env table expr =
  let module SS = Constraint.StringSet in
  let table_name = Sql.join_source_name table in
  let belongs (a : table_name Schema.Source.Attr.t) =
    List.exists (fun (s : table_name) -> s.tn = table_name.tn) a.sources
  in
  let table_attrs =
    List.filter_map
      (fun a -> if belongs a then Some a.Schema.Source.Attr.attr else None)
      env.schema
  in
  let keys = unique_keys table_attrs in
  let independent_of_table e =
    not (Table_refs.may_refer table (Table_refs.of_expr ~env e))
  in
  let as_column = function
    | Sql.Column c -> resolve_column_opt ~env c.collated
    | _ -> None
  in
  let bound1 a b =
    match as_column a with
    | Some attr when belongs attr && independent_of_table b ->
      Some attr.Schema.Source.Attr.attr.name
    | _ -> None
  in
  let bound_part a b = match bound1 a b with Some _ as r -> r | None -> bound1 b a in
  let rec bound_parts = function
    | Sql.Fun { kind = Logical And; parameters; _ } ->
      List.fold_left (fun acc e -> SS.union acc (bound_parts e)) SS.empty parameters
    | Fun { kind = Comparison Comp_equal; parameters = [a; b]; _ } ->
      b |> bound_part a |> Option.map_default SS.singleton SS.empty
    | Choices (_, branches) ->
      let of_branch (_, e) = Option.map_default bound_parts SS.empty e in
      (match branches with
       | [] -> SS.empty
       | hd :: tl -> List.fold_left (fun acc b -> SS.inter acc (of_branch b)) (of_branch hd) tl)
    | Fun _ | Value _ | Param _ | Inparam _ | Column _ | Of_values _ | SelectExpr _
    | InChoice _ | OptionActions _ | InTupleList _ | Case _ -> SS.empty
  in
  let bound = bound_parts expr in
  List.exists (fun k -> SS.subset k bound) keys

module Table_elimination = struct

  module Id_set = Set.Make(Int)
  module Id_map = Map.Make(Int)
  module Table_map = Map.Make(String)

  type candidate = {
    table : Sql.join_source;
    join : From.join;
  }

  let join_id c = fst c.join.pos

  let eliminate ~env ~from ~columns ~where ~group ~having ~order final_schema from_params =
    let unchanged = final_schema, from_params in
    let joins = Option.map_default (fun f -> f.From.joins) [] from in
    let eliminable ({ From.src; kind; cond; pos = _ } as join) =
      let has_params = expr_exists (function
        | Sql.Param _ | Inparam _ | InTupleList _ | Choices _ | InChoice _
        | OptionActions _ | SelectExpr _ -> true
        | Value _ | Column _ | Of_values _ | Fun _ | Case _ -> false)
      in
      match kind, cond, src.rsrc_physical_table with
      | Schema.Join.Left, Schema.Join.On e, Some table
        when not (has_params e) && matches_at_most_one_row ~env table e ->
        Some { table; join }
      | _ -> None
    in
    let is_implicit j = match j.From.cond with
      | Schema.Join.Natural | Using _ -> true
      | On _ | Default -> false
    in
    let rec after_last_implicit l =
      match List.dropwhile (not $ is_implicit) l with
      | [] -> l
      | _ :: rest -> after_last_implicit rest
    in
    let candidates =
      joins
      |> after_last_implicit
      |> List.filter_map eliminable
      |> List.fold_left (fun m c -> Id_map.add (join_id c) c m) Id_map.empty
    in
    if Id_map.is_empty candidates then unchanged else
    let outside_select_list = option_list where @ group @ option_list having @ List.map fst order in
    let query_exprs =
      List.filter_map (fun c -> match c.Sql.value with
        | All | AllOf _ -> None
        | Expr ({ value = e; _ }, _) -> Some e)
        columns
      @ outside_select_list
    in
    if List.exists is_windowing query_exprs then unchanged else
    let keys_where p m = Id_map.fold (fun k v acc -> if p k v then Id_set.add k acc else acc) m Id_set.empty in
    let used_elsewhere =
      let static_select_list =
        List.concat_map (fun c -> match c.Sql.value with
          | All | AllOf _ -> []
          | Expr ({ value = Choices (_, choices); _ }, _) ->
            List.filter_map (function
              | (_, Some (Sql.Column _)) | (_, None) -> None
              | (_, Some e) -> Some e) choices
          | Expr ({ value = Column _; _ }, _) -> []
          | Expr ({ value = e; _ }, _) -> [e])
          columns
      in
      Table_refs.of_exprs ~env (outside_select_list @ static_select_list)
    in
    let condition_refs =
      List.fold_left (fun m { From.cond; pos; _ } ->
        match cond with
        | Schema.Join.On e ->
          let refs = Table_refs.of_expr ~env e in
          let j = fst pos in
          let referenced =
            candidates
            |> keys_where (fun _ c -> Table_refs.may_refer c.table refs)
            |> Id_set.remove j
          in
          Id_map.add j referenced m
        | Default | Natural | Using _ -> m)
        Id_map.empty joins
    in
    let condition_refs_of j = condition_refs |> Id_map.find_opt j |> Option.default Id_set.empty in
    let saturate refs set =
      let rec go s =
        let expanded = Id_set.fold (fun j -> Id_set.union (refs j)) s s in
        if Id_set.equal expanded s then s else go expanded
      in
      go set
    in
    let redundant_ids =
      let unreferenced =
        keys_where (fun _ c -> not (Table_refs.may_refer c.table used_elsewhere)) candidates
      in
      let retained = Id_set.diff (keys_where (fun _ _ -> true) condition_refs) unreferenced in
      Id_set.diff unreferenced (saturate condition_refs_of retained)
    in
    let direct i = Id_set.inter (condition_refs_of i) redundant_ids in
    let by_table =
      Id_set.fold (fun j m ->
        let tn = (Sql.join_source_name (Id_map.find j candidates).table).tn in
        Table_map.update tn (fun old -> Some (Id_set.add j (Option.default Id_set.empty old))) m)
        redundant_ids Table_map.empty
      |> Table_map.map (saturate direct)
    in
    let join_of_column a =
      List.find_map (fun s -> Table_map.find_opt s.tn by_table) a.Schema.Source.Attr.sources
    in
    let annotate_column needed field =
      match join_of_column field.Sql.field_attr with
      | None -> needed, field
      | Some pre ->
        Id_set.union needed pre,
        { field with Sql.join_deps = Id_set.elements pre }
    in
    let pid =
      List.find_map (function
        | DynamicWithSources (p, _) -> Some p
        | AttrWithSources _ -> None) final_schema
    in
    let needed, final_schema =
      List.fold_left_map (fun needed -> function
        | DynamicWithSources (p, cols) ->
          let needed, cols = List.fold_left_map annotate_column needed cols in
          needed, DynamicWithSources (p, cols)
        | AttrWithSources _ as x -> needed, x)
        Id_set.empty final_schema
    in
    let holes = match pid with
    | None -> []
    | Some pid ->
      needed
      |> Id_set.elements
      |> List.map (fun j ->
        let c = Id_map.find j candidates in
        Sql.DynamicSelectJoin { pid; pos = c.join.pos; source = c.table })
    in
    let by_position a b = Int.compare (Sql.var_pos a) (Sql.var_pos b) in
    final_schema, List.merge by_position from_params (List.sort ~cmp:by_position holes)
end

(** resolve each name reference (Column, Inserted, etc) into ResValue or ResFun of corresponding type *)
let rec resolve_columns env expr =
  if !Config.debug then
  begin
    eprintf "\nRESOLVE COLUMNS %s\n%!" (expr_to_string expr);
    eprintf "schema: "; Sql.Schema.print (Schema.Source.from_schema env.schema);
    Tables.print stderr env.tables;
  end;
  let expr = extract_meta_from_col ~env expr in
  let rec each e =
    match e with
    | Value x -> ResValue x.collated
    | Column col ->
      let attr = (resolve_column ~env col.collated).attr in
      let json_null_kind = Meta.find_opt attr.meta "json_null_kind" in
      let text_as_json = Meta.find_opt attr.meta "text_as_json" in
      let domain = match json_null_kind, text_as_json, attr.domain with
        | v, _, ({ t = Json; nullability } as d)
        | v, Some "true", ({ t = Text; nullability } as d) -> 
          (*
            Determines whether JSON null is allowed as a valid value in the column.

            JSON null (i.e. the literal `null` in a JSON document) is distinct from SQL NULL.
            - JSON null is an actual value in JSON and can appear inside arrays or objects.
            - SQL NULL means "no value at all" and causes most JSON functions to return NULL
              if encountered as an argument (e.g. JSON_ARRAY_APPEND returns NULL if any argument is SQL NULL).

            Standard SQL DDL (e.g. CREATE TABLE) does not allow expressing whether JSON null is allowed
            for JSON columns — it only covers SQL NULL via NOT NULL constraints.

            To bridge this semantic gap, we introduce a custom meta-attribute `json_null_kind`:
              - "true" or "auto" → JSON null is allowed (treated as a Nullable domain)
              - "false"          → JSON null is disallowed (treated as Strict)

            Additionally, if `text_as_json` is set and the underlying type is `Text`,
            we apply the same logic (since JSON is serialized into text in that case).

            The resulting domain is:
              - Nullable → JSON null is allowed in values
              - Strict   → JSON null is rejected during validation

            This impacts how JSON expressions are parsed, validated, and how DDL is generated.
          *)
          let nullability = match v, nullability with 
          | Some "false", Type.Strict -> Type.Strict
          | _ -> Type.Nullable
          in
          { Type.t = d.t; nullability; }
        | _, Some _, _ -> 
          fail "Column %s has text_as_json meta, but its type is not Text" col.collated.cname  
        | Some _, _, _ -> 
          fail "Column %s has json_null_kind meta, but its type is not Json or Text" col.collated.cname
        | None, _, _ -> 
          attr.domain
      in
      ResValue domain
    | OptionActions { choice; pos; kind } ->
      let choice_id = match bool_choice_id choice with
      | Some choice_id -> choice_id
      | None -> 
        fail "BoolChoices expected a parameter, but isn't presented. Use regular Choices for this kind of logic"
      in
      ResOptionActions { res_choice = each choice; choice_id; pos; kind }
    | Param (x, m) -> ResParam (make_param ~id:x.id ~typ:(Source_type.to_infer_type x.typ), m)
    | InTupleList ({ value = { exprs; param_id; kind_in_tuple_list; }; pos } ) -> 
      let res_exprs = List.map (fun expr ->
        let res_expr = each expr in
        match res_expr with 
        | ResCase _
        | ResValue _
        | ResParam _
        | ResSelect _
        | ResFun _ -> res_expr
        | ResInparam _
        | ResChoices _
        | ResInTupleList _
        | ResOptionActions _
        | ResInChoice _ -> fail "unsupported expression %s kind for WHERE e IN @tuplelist" (show_res_expr res_expr)
      ) exprs in
      let res_exprs = List.map2 (fun e re ->
        match e with
        | Column col -> re, (resolve_column ~env col.collated).attr.meta 
        | _ -> re, Meta.empty ()
      ) exprs res_exprs in
      ResInTupleList {param_id; res_in_tuple_list = Res res_exprs; kind = kind_in_tuple_list; pos }
    | Inparam (x, m) -> ResInparam (make_param ~id:x.id ~typ:(Source_type.to_infer_type x.typ), m)
    | InChoice (n, k, x) -> ResInChoice (n, k, each x)
    | Choices (n, l) -> ResChoices (n, List.map (fun (n, e) -> n, Option.map each e) l)
    | Fun { kind; parameters; is_over_clause; _ } ->
      ResFun { kind = source_fun_kind_to_infer kind; parameters = List.map each parameters; is_over_clause }
    | Case { case; branches; else_ } ->
      let case = Option.map each case in
      let branches = List.map (fun { Sql.when_; then_ } -> { when_ = each when_; then_ = each then_ }) branches in
      let else_ = Option.map each else_ in
      ResCase { case; branches; else_ }
    | Of_values col -> begin match Hashtbl.find_opt env.insert_resolved_types col with
      | Some t -> ResValue t
      | None -> fail "VALUES(col) as an expression is only acceptable in ON DUPLICATE KEY UPDATE context" 
      end
    (* nested select *)
    | SelectExpr (select, usage) ->
      let (schema, p, _) = eval_select_full { env with scope = Subquery } select in
      let schema = List.map (function
        | AttrWithSources a -> a
        | DynamicWithSources _ -> fail "nested select cannot have dynamic attributes"
      ) schema
      in
      let schema' = Schema.Source.from_schema schema in
      (* represet nested selects as functions with sql parameters as function arguments, some hack *)
      match schema, usage with
      | [ { attr = {domain; _}; _ } ], `AsValue -> 
        (* This function should be raised? *)
        let rec with_count = function 
            | Case { case = _; branches; else_ } ->
              let then_exprs = List.map (fun b -> b.Sql.then_) branches in
              let all_results_exprs = then_exprs @ (option_list else_) in
              List.find_map with_count all_results_exprs
            | Fun { kind = Agg Count; is_over_clause = false; _ }
            | SelectExpr (_, _) -> Some domain
            | Fun { parameters; is_over_clause = false; _ } -> List.find_map with_count parameters
            | Choices (_, chs) ->
              List.fold_left (fun acc (_, e) -> match acc with
                | None -> None
                | Some _ -> Option.map_default with_count None e
              ) (Some domain) chs
            | OptionActions { choice; _ } -> with_count choice  
            | Fun { is_over_clause = true; _ }
            | Value _| Param _| Inparam _ | InChoice _
            | Column _| InTupleList _ | Of_values _ -> None
        in
        let default_null = Type.make_nullable domain in
        (* The only way to have a result in a subquery is to use the COUNT function wihout the HAVING expression. 
           Any other expression could possibly return no rows. *)
        let typ = match select.select_complete.select with 
        | ({ having = Some _; _ }, _) -> Type.nullable domain.t
        | ({ columns = [{ value = Expr ({ value = c; _ }, _); _ }]; _ }, _) -> c |> with_count |> Option.default default_null
        | ({ columns = [_]; _ }, _) -> default_null
        | _ -> raise (Schema.Error (schema', "nested sub-select used as an expression returns more than one column"))
        in
        ResSelect (typ, p)
      | _, `AsValue -> raise (Schema.Error (schema', "only one column allowed for SELECT operator in this expression"))
      | _, `Exists -> ResSelect (Type.depends Any, p)
  in
  each expr

(** assign types to parameters where possible *)
and assign_types env expr =
  let { set_tyvar_strict; _ } = env in
  let option_split = function None -> None, None | Some (x,y) -> Some x, Some y in
  let assign_params inferred x =
    match x with
    | ResParam ({ id; typ; }, m) when Type.is_any typ -> ResParam (make_param ~id ~typ:inferred, m)
    | ResInparam ({ id; typ; }, m) when Type.is_any typ -> ResInparam (make_param ~id ~typ:inferred, m)
    | x -> x
  in
  let rec typeof_ (e:res_expr) = (* FIXME simplify *)
    match e with
    | ResValue t -> e, `Ok t
    | ResParam (p, _) -> e, `Ok p.typ
    | ResInparam (p, _) -> e, `Ok p.typ
    | ResSelect (t, _) -> e, `Ok t
    | ResOptionActions choice ->
      let (res_choice, t) = typeof choice.res_choice in
      let t =
        match Type.common_subtype [Type.depends Bool; get_or_failwith t] with
        | None -> `Error "no common subtype for ResOptionActions"
        | Some t -> `Ok t
      in
      ResOptionActions { choice with res_choice }, t
    | ResInTupleList { param_id; res_in_tuple_list; kind; pos } -> 
      (match res_in_tuple_list with 
      | Res res_exprs -> ResInTupleList { param_id; 
        res_in_tuple_list = ResTyped (List.map (fun (expr, meta) ->
          let typ = expr |> typeof |> snd |> get_or_failwith in 
          if Type.is_any typ then 
              fail "If you need to have a field as parameter in the left part you should specify a type"
          else typ, meta
        ) res_exprs); kind; pos }, `Ok (Type.strict Bool) 
      | ResTyped _ -> assert false
      )
    | ResInChoice (n, k, e) -> let e, t = typeof e in ResInChoice (n, k, e), t
    | ResChoices (n,l) ->
      let (e,t) = List.split @@ List.map (fun (_,e) -> option_split @@ Option.map typeof e) l in
      let t =
        match Type.common_subtype @@ List.map get_or_failwith @@ List.filter_map identity t with
        | None -> `Error "no common subtype for all choice branches"
        | Some t -> `Ok t
      in
      (* We can order by different columns with the different types *)
      let assign_any e = if env.is_order_by then e else assign_params (get_or_failwith t) e in
      ResChoices (n, List.map2 (fun (n,_) e -> n, (Option.map assign_any e)) l e), t
    | ResCase { case; branches; else_ } ->
      let (case_e, case_t) = option_split @@ Option.map typeof case in
      let (else_, else_t) = option_split @@ Option.map typeof else_ in
      let (whens_e, whens_t) = List.split @@ List.map (fun { when_; _ } -> typeof when_) branches in
      let (thens_e, thens_t) = List.split @@ List.map (fun { then_; _} -> typeof then_) branches in
      let whens_t =
        let types = List.map get_or_failwith @@ whens_t in
        match Type.common_supertype @@ Option.map_default get_or_failwith (Type.depends Bool) case_t :: types with
        | None -> failwith "no common supertype for all case when branches"
        | Some t -> t
      in
      let thens_t =
        let types = List.map get_or_failwith @@ thens_t in
        let is_exhausted = match whens_t.t with
          | Union { ctors; _ } -> 
            (* Since we have string literals, we can check if the enums are already exhausted or if a default case is required *)
            let values = Type.Enum_kind.Ctors.of_list @@ List.filter_map (function ResValue { t = StringLiteral v; _ } -> Some v | _ -> None) whens_e in
            Type.Enum_kind.Ctors.compare values ctors = 0
          | Int | UInt64 | Text | Blob | Float | Datetime | FloatingLiteral _
          | Decimal _ | Any | One_or_all | Json | StringLiteral _ | Json_path | Bool -> false in
        let exhaust_checked = if is_exhausted then types else List.map Type.make_nullable types in  
        match Type.common_supertype @@ Option.map_default (fun else_t -> types @ [get_or_failwith else_t]) exhaust_checked else_t with
        | None -> failwith "no common supertype for all case then branches"
        | Some t -> t
      in
      let thens_e = List.map (assign_params thens_t) thens_e in
      let whens_e = List.map (assign_params whens_t) whens_e in
      let else_ = Option.map (assign_params thens_t) else_ in
      let case = Option.map (assign_params whens_t) case_e in
      let branches = List.map2 (fun when_ then_ -> { when_; then_ }) whens_e thens_e in
      ResCase { case = case; branches; else_ = else_ }, `Ok thens_t
    | ResFun { kind; parameters; is_over_clause}  ->
        let open Type in
        let (params,types) = parameters |> List.map typeof |> List.split in
        let types = List.map get_or_failwith types in
        let show_func () =
          sprintf "%s applied to (%s)"
            (string_of_func kind)
            (String.concat ", " @@ List.map show types)
        in
        if !Config.debug then eprintfn "func %s" (show_func ());
        let types_to_arg each_arg = List.map (const each_arg) types in
        let convert_args ret args = 
          let typevar = Hashtbl.create 10 in
          let resolved_typs = Hashtbl.create 10 in

          List.iteri (fun idx arg ->
            let typ = List.nth types idx in
            match arg with
            | Typ arg_ty ->
                begin match common_type arg_ty typ with
                | Some unified -> Hashtbl.add resolved_typs idx unified
                | None -> fail "types %s and %s do not match in %s" (show arg_ty) (show typ) (show_func ())
                end
            | Var i ->
                let var = Hashtbl.find_default typevar i typ in
                begin match common_type var typ with
                | Some t ->
                    if !Config.debug then Type.(eprintfn "common_type %s %s = %s" (show var) (show typ) (show t));
                    Hashtbl.replace typevar i t
                | None ->
                    fail "types %s and %s for %s do not match in %s"
                      (show var) (show typ) (string_of_tyvar arg) (show_func ())
                end
          ) args;

          if !Config.debug then
            Hashtbl.iter (fun i typ -> eprintfn "%s : %s" (string_of_tyvar (Var i)) (show typ)) typevar;

          let resolve_arg idx = function
            | Typ t -> Hashtbl.find_default resolved_typs idx t
            | Var i -> Hashtbl.find typevar i
          in
          let args = List.mapi resolve_arg args in
          let ret = match ret with
            | Typ t -> t
            | Var i -> Hashtbl.find typevar i
          in
          args, ret in

        (* With GROUP BY, the query returns no rows if no groups exist. With OVER clause, the query returns no rows if the outer query filter eliminates all rows.
           In both cases, if we're in a context that expects a value (like a subquery), the result should be nullable. *)
        let consider_agg_nullability typ = if (env.query_has_grouping || is_over_clause) && is_strict typ then typ else make_nullable typ in

        let first_strict ret args = 
          let has_one_strict = List.exists (fun arg -> equal_nullability arg.nullability Strict) types in
          let ret = if has_one_strict then make_strict ret
            else args |> common_nullability |> undepend ret in 
          ret , args
        in

        let rec infer_fn func types = match func, types with
        | Multi { ret; fixed_args = []; repeating_pattern = [each_arg] }, t -> 
          infer_fn (F (ret, types_to_arg each_arg)) t
        | Multi { ret; fixed_args; repeating_pattern }, t->
          let fixed_count = List.length fixed_args in
          let pattern_length = List.length repeating_pattern in
          let total_args = List.length types in

          if total_args < fixed_count then
            fail "function %s requires at least %d arguments, got %d" 
              (show_func ()) fixed_count total_args
          else if Stdlib.(pattern_length = 0) then (
            if total_args <> fixed_count then
              fail "function %s requires exactly %d arguments, got %d" 
          (show_func ()) fixed_count total_args
            else
              infer_fn (F (ret, fixed_args)) t
          ) else if (total_args - fixed_count) mod pattern_length <> 0 then
            fail "function %s requires %d fixed args + multiples of %d args, got %d" 
              (show_func ()) fixed_count pattern_length total_args
          else
            let remaining_count = total_args - fixed_count in
            let repeating_cycles = remaining_count / pattern_length in
            let repeated_args =
              List.flatten (List.init repeating_cycles (Fun.const repeating_pattern))
            in
            infer_fn (F (ret, fixed_args @ repeated_args)) t
        | Agg Count, ([] (* asterisk *) | [_]) -> strict Int, types
        | Agg Avg, [_] -> consider_agg_nullability @@ nullable Float, types
        | Agg Self, [typ] -> consider_agg_nullability typ, types
        | Agg (With_order { with_order_kind = Group_concat; _ }), ((_ :: _) as params) -> 
          let ret = depends Text in
          let nullability = common_nullability (ret :: params) in
          consider_agg_nullability @@ (undepend ret nullability), types
        | Agg (With_order { with_order_kind = Json_arrayagg; _ }), [t1] -> 
          let ret = depends Json in
          let nullability = common_nullability [ret; t1] in
          consider_agg_nullability @@ (undepend ret nullability), types
        | Agg _, _ -> fail "cannot use this grouping function with %d parameters" (List.length types)
        | F (_, args), _ when List.length args <> List.length types -> fail "wrong number of arguments : %s" (show_func ())
        | Null_handling (Coalesce (ret, each_arg)) , _ -> 
          let args = types_to_arg each_arg in
          let args, ret = convert_args ret args in
          first_strict ret args
        | Null_handling Null_if , _ ->
          let args, ret = convert_args (Var 0) [Var 0; Var 0] in
          make_nullable ret, args
         | Null_handling If_null, _ ->
          let args, ret = convert_args (Var 0) [Var 0; Var 0] in
          first_strict ret args
        | F (ret, args), _ ->
          let args, ret = convert_args ret args in
          let nullable = common_nullability args in
          undepend ret nullable, args
        | Ret t, _ when Type.is_any t -> (* lame *)
          begin match common_supertype types with
          | Some t -> t, List.map (fun _ -> t) types
          | None -> { t = Any; nullability = common_nullability types }, types
          end
        | Ret ret, _ ->
          let nullability = common_nullability @@ ret :: types in (* remove this when subqueries are taken out separately *)
          { ret with nullability }, types (* ignoring arguments FIXME *)
        | Comparison Not_distinct_op, _ ->
          let args, ret = convert_args (Typ (strict Bool)) [Var 0; Var 0] in
          ret, args
        | Comparison (Is_not_null | Is_null), _ ->
          let args, ret = convert_args (Typ (strict Bool)) [Var 0] in
          ret, args
        | Comparison _, _ when set_tyvar_strict ->
        (* In this expression, where set_tyvar_strict is set (currently only for WHERE) we treat the parameters as non-null by default. *)
          let args, ret = convert_args (Typ (depends Bool)) [Var 0; Var 0] in
          let strict_args = List.map2 (fun param_expr inferred_type ->
            match param_expr with
            | ResParam _ -> make_strict inferred_type
            | _ -> inferred_type
          ) parameters args in
          ret, strict_args
        | Comparison _, _  ->
          let args, ret = convert_args (Typ (depends Bool)) [Var 0; Var 0] in
          let nullable = common_nullability args in
          undepend ret nullable, args
        | Negation, [_] -> 
          infer_fn (fixed Bool [Bool]) types
        | Negation, _ -> fail "negation requires a single argument"
        | Logical _, [_ ; _] ->
          infer_fn (fixed Bool [Bool; Bool]) types
        | Logical _, _ -> fail "logical operators require two arguments"
        | Col_assign { ret_t; col_t; arg_t }, [a; b] ->
          let args, ret = convert_args ret_t [col_t; arg_t] in
          let t =
            if !Config.allow_write_notnull_null && Dialect.Semantic.is_non_strict_mode_is_exists() then 
            undepend ret (common_nullability args)
            else 
            let nullability = match order_nullability a.nullability b.nullability with
              | `Equal n -> n
              | `Nullable_Strict -> b.nullability
              | `Strict_Nullable -> fail "Cannot assign nullable value to a non-nullable column %s" (show_func ())
            in 
            { ret with nullability }
          in
          t, args
        | Col_assign _, _ -> fail "SET operation requires two arguments"
        in
        let (ret,inferred_params) = infer_fn kind types in
        ResFun { kind; parameters = (List.map2 assign_params inferred_params params); is_over_clause }, `Ok ret
  and typeof expr =
    let r = typeof_ expr in
    if !Config.debug then eprintfn "%s is typeof %s" (Type.show @@ get_or_failwith @@ snd r) (show_res_expr @@ fst r);
    r
  in
  typeof expr

and resolve_types env expr =
  let expr = expr |> resolve_columns env in
  try
    assign_types env expr
  with
    exn ->
      if !Config.debug then begin
        eprintfn "resolve_types failed with %s at:" (Printexc.to_string exn);
        eprintfn "%s" (show_res_expr expr)
      end;
      raise exn

and infer_schema ~not_null_keys env columns =
(*   let all = tables |> List.map snd |> List.flatten in *)
  let rec propagate_meta ~env = function
    | Column col -> 
      let result = resolve_column ~env col.collated in 
      result.attr.meta
    (* aggregated columns, ie: max, min *)
    | Fun { kind = Agg Self; parameters = [e]; _ } -> propagate_meta ~env e
     (* null handling functions that preserve metadata from first argument *)
    | Fun { kind = Null_handling (Coalesce _ | If_null); parameters = e :: _; _ } -> propagate_meta ~env e
    (* Or for subselect which always requests only one column, TODO: consider CTE in subselect, perhaps a rare occurrence *)
    | SelectExpr ({ select_complete = { select = ({columns = [{ value = Expr ({ value; _ }, _); _ }]; from; _}, _); _ }; _ }, _) ->
      let (env,_,_) = eval_nested { env with scope = Subquery } from in
      propagate_meta ~env value
    | Case _
    | Value _
    (* TODO: implement for custom props *)
    | Param _ | Inparam _ | Choices _| InChoice _
    | Fun _ | SelectExpr _ | InTupleList _ | Of_values _
    | OptionActions _ -> Meta.empty ()
  in
  let refine_column (col : table_name Schema.Source.Attr.t) =
    if List.mem (col.sources, col.attr.name) not_null_keys
    then Schema.Source.Attr.map_attr (fun attr -> { attr with domain = Type.make_strict attr.domain }) col
    else col
  in
  let resolve1 = function
    | { value = All; _ } -> List.map (fun x -> AttrWithSources (refine_column x)) env.schema
    | { value = AllOf t; _ } -> List.map (fun x -> AttrWithSources (refine_column x)) (schema_of ~env t)
    | { value = Expr ({ value = expr; _ }, alias); _ } ->
      let apply_alias col =
        Option.map_default
          (fun n -> Schema.Source.Attr.map_attr (fun attr -> { attr with name = n }) col)
          col alias
      in
      let resolve_expr = function
        | Column c -> resolve_column ~env c.collated
        | e ->
          let _, t = resolve_types env e in
          { Schema.Source.Attr.attr = unnamed_attribute ~meta:(propagate_meta ~env e) (get_or_failwith t);
            sources = [] }
      in
      let col =
        match expr with
        | Choices (p, choices) when dynamic_allowed env ->
          let dynamic = choices |> List.filter_map (fun (choice_p, e_opt) -> 
            Option.map (fun choice_e ->
              let field_attr =
                choice_e
                |> resolve_expr |> refine_column
                |> Schema.Source.Attr.map_attr (fun attr -> unnamed_attribute ~meta:attr.meta attr.domain)
                |> apply_alias
              in
              { Sql.field_id = choice_p; field_attr; join_deps = [] }) e_opt
          ) in 
          DynamicWithSources (p, dynamic)
        | e -> AttrWithSources (e |> resolve_expr |> refine_column |> apply_alias)
      in
      [ col ]
  in
  flat_map resolve1 columns

and get_params env e = e |> resolve_types env |> fst |> get_params_of_res_expr env

(*
let _ =
  let e = Sub [Value Type.Text; Param (Next,None); Sub []; Param (Named "ds", Some Type.Int);] in
  e |> get_params |> to_string |> print_endline
*)

and get_params_of_columns env =
  let get = function
  | { value = (All | AllOf _); _ } -> []
  | { value = Expr ({ value = Choices (p, choices); _ }, _); _ } when dynamic_allowed env ->
    [DynamicSelect (p, List.map (fun ((n : param_id), e) -> 
      match e with
      | Some (Column { collated = { cname; tname }; _ }) when n.pos = dummy_pos ->
        let sql = tname |> Option.map_default (fun t -> Printf.sprintf "%s.%s" (show_table_name t) cname) cname in
        Verbatim (Option.default cname n.value, sql)
      | _ ->
        Simple (n, Option.map (fun e -> e |> resolve_types env |> fst |> get_params_of_res_expr env) e)
    ) choices)]
  | { value = Expr ({ value; _ }, _); _ } -> get_params env value
  in
  flat_map get

and get_params_opt env = function
  | Some x -> get_params env x
  | None -> []

and get_params_l env l = flat_map (get_params env) l

and do_join (env,params) { From.src; kind; cond; _ } =
  let schema = Schema.Join.join kind cond env.schema src.rsrc_schema in
  let env = { env with schema } in
  let p = match cond with
  | Default | Natural | Using _ -> []
  | On e -> get_params { env with set_tyvar_strict = true } e (* TODO should use final schema (same as tables)? *)
  in
  env, params @ src.rsrc_params @ p

and join env { From.base; joins } =
  assert (env.schema = []);
  let all_tables = base.rsrc_tables @ List.concat_map (fun j -> j.From.src.rsrc_tables) joins in
  let env = { env with tables = env.tables @ all_tables; schema = base.rsrc_schema } in
  List.fold_left do_join (env, base.rsrc_params) joins


and params_of_assigns env ss =
  let exprs = resolve_column_assignments ~env ss in
  get_params_l env exprs

and get_params_of_res_expr env e =
  let rec loop acc e =
    match e with
    | ResSelect (_, p) -> (List.rev p) @ acc
    | ResCase { case; branches; else_ } ->
      let acc = match case with Some e -> loop acc e | None -> acc in
      let acc = List.fold_left (fun acc { when_; then_ } -> loop (loop acc when_) then_) acc branches in
      Option.map_default (loop acc) acc else_
    | ResParam (p, m) -> Single (p, m) ::acc
    | ResOptionActions{ choice_id; res_choice; pos; kind} -> 
      OptionActionChoice (choice_id, get_params_of_res_expr env res_choice, pos, kind) :: acc
    | ResInTupleList { param_id; res_in_tuple_list = ResTyped types; kind; pos } -> TupleList (param_id, Where_in { value = (types, kind); pos }) :: acc
    | ResInparam (p, m) -> SingleIn (p, m)::acc
    | ResFun { parameters; kind; _ } -> 
      let p1 = match kind with
      | Agg (With_order { order; _ }) -> List.rev @@ params_of_order order [] env
      | _ -> [] in
      p1 @ List.fold_left loop acc parameters
    | ResInTupleList _
    | ResValue _ -> acc
    | ResInChoice (param, kind, e) -> ChoiceIn { param; kind; vars = get_params_of_res_expr env e } :: acc
    | ResChoices (p, l) -> Choice (p, List.map (fun (n, e) -> Simple (n, Option.map (get_params_of_res_expr env) e)) l) :: acc
  in
  loop [] e |> List.rev

and params_of_order order final_schema env =
  List.concat_map
    (fun (order, direction) ->
       let env = { env with schema = update_schema_with_aliases env.schema final_schema ;  } in
       let p1 = get_params_l { env with is_order_by = true } [ order ] in
       let p2 =
         match direction with
         | None | Some `Fixed -> []
         | Some (`Param p) -> [Choice (p,[Verbatim ("ASC","ASC");Verbatim ("DESC","DESC")])]
       in
       p1 @ p2)
    order

and ensure_res_expr = function
  | Value x -> ResValue x.collated
  | Param (x, m) -> ResParam (make_param ~id:x.id ~typ:(Source_type.to_infer_type x.typ), m)
  | Inparam (x, m) -> ResInparam (make_param ~id:x.id ~typ:(Source_type.to_infer_type x.typ), m)
  | Case { case; branches; else_ }-> 
    let res_case = Option.map ensure_res_expr case in
    let res_branches = List.map (fun { Sql.when_; then_ } -> 
      { when_ = ensure_res_expr when_; then_ = ensure_res_expr then_ }
    ) branches in
    let res_else = Option.map ensure_res_expr else_ in
    ResCase { case = res_case; branches = res_branches; else_ = res_else }
  | InTupleList { value = { param_id; _ }; _ } -> failed ~at:param_id.pos "ensure_res_expr InTupleList TBD"
  | Choices (p,_) -> failed ~at:p.pos "ensure_res_expr Choices TBD"
  | InChoice (p,_,_) -> failed ~at:p.pos "ensure_res_expr InChoice TBD"
  | Column _ | Of_values _ -> failwith "Not a simple expression"
  | Fun { kind; _ } when Sql.is_grouping kind -> failwith "Grouping function not allowed in simple expression"
  | Fun { kind; parameters; is_over_clause; _ } ->
     ResFun { kind = source_fun_kind_to_infer kind; parameters = List.map ensure_res_expr parameters; is_over_clause } (* FIXME *)
  | SelectExpr _ -> failwith "not implemented : ensure_res_expr for SELECT"
  | OptionActions _ -> failwith  "BoolChoice is used in WHERE expr only"

and eval_nested env nested =
  (* nested selects generate new fresh schema in scope, cannot refer to outer schema,
    but can refer to attributes of tables through `tables` *)
  let env = { env with schema = [] } in
  (* FIXME resolved table schema depends on join (nullability with left), this is resolving too early *)
  match nested with
  | Some (t,l) ->
    let resolve = resolve_source env in
    let from = {
      From.base = resolve t;
      joins = List.map (fun loc ->
        let (x,jt,jc) = loc.value in
        { From.src = resolve x; kind = jt.value; cond = jc; pos = loc.pos }) l;
    } in
    let env, params = join env from in
    env, params, Some from
  | None -> env, [], None

(** Extract (sources, name) pairs for columns with IS NOT NULL in WHERE/HAVING

    Uses structured nullability analysis with uniform logic rules:
    - Builds nullability AST: IsNotNull, IsNull, And, Or, Not
    - Applies De Morgan's laws automatically
    - Handles all combinations: NOT (A AND B), NOT (A OR B), nested logic
*)
and extract_not_null_column_keys env = function
  | None -> []
  | Some expr ->
    let module NullCheck = struct
      (* Nullability analysis AST *)
      type t =
        | IsNotNull of (table_name list * string)
        | IsNull of (table_name list * string)
        | And of t list
        | Or of t list
        | Unknown

      (* Apply De Morgan's laws uniformly *)
      let rec negate = function
        | IsNotNull col -> IsNull col
        | IsNull col -> IsNotNull col
        | And checks -> Or (List.map negate checks)
        | Or checks -> And (List.map negate checks)
        | Unknown -> Unknown
    end in

    (* Build nullability AST from expression *)
    let rec analyze = function
      | Column _ -> NullCheck.Unknown
      | Fun { kind = Comparison Is_not_null; parameters = [Column col]; _ } ->
          let resolved = resolve_column ~env col.collated in
          NullCheck.IsNotNull (resolved.sources, resolved.attr.name)
      | Fun { kind = Comparison Is_null; parameters = [Column col]; _ } ->
          let resolved = resolve_column ~env col.collated in
          NullCheck.IsNull (resolved.sources, resolved.attr.name)
      | Fun { kind = Negation; parameters = [e]; _ } ->
          NullCheck.negate (analyze e)
      | Fun { kind = Logical And; parameters; _ } ->
          NullCheck.And (List.map analyze parameters)
      | Fun { kind = Logical Or; parameters; _ } ->
          NullCheck.Or (List.map analyze parameters)
      | Fun _ -> NullCheck.Unknown
      | Case { case = _; branches; else_ } ->
          (* CASE in WHERE: all THEN branches + ELSE must be analyzed *)
          let then_checks = List.map (fun { Sql.then_; _ } -> analyze then_) branches in
          let else_check = Option.map_default analyze NullCheck.Unknown else_ in
          begin match else_ with
          | None -> NullCheck.Unknown  (* No ELSE = can't guarantee all paths *)
          | Some _ -> NullCheck.Or (then_checks @ [else_check])  (* One of them will be true *)
          end
      | Choices (_, choices) ->
          (* User chooses one branch at runtime *)
          let branch_checks = List.map (fun (_pid, e_opt) ->
            Option.map_default analyze NullCheck.Unknown e_opt
          ) choices in
          NullCheck.Or branch_checks  (* One of them will be chosen *)
      | InChoice (_, _, e) -> analyze e
      | OptionActions { choice; _ } -> analyze choice
      | SelectExpr _ | Value _ | Param _ | Inparam _
      | InTupleList _ | Of_values _ -> NullCheck.Unknown
    in

    (* Extract columns guaranteed to be non-null from nullability AST *)
    let rec extract = function
      | NullCheck.IsNotNull col -> [col]
      | NullCheck.IsNull _ -> []
      | NullCheck.And checks ->
          (* All columns that are IsNotNull in ANY branch of AND *)
          List.concat_map extract checks
      | NullCheck.Or checks ->
          (* Only columns that are IsNotNull in ALL branches of OR *)
          let all_lists = List.map extract checks in
          let all_cols = List.concat all_lists in
          List.filter (fun col ->
            List.for_all (fun branch -> List.mem col branch) all_lists
          ) (List.sort_uniq compare all_cols)
      | NullCheck.Unknown -> []
    in

    expr |> analyze |> extract

and eval_select ~order env { columns; from; where; group; having; } =
  let is_passthrough = columns <> [] && List.for_all (fun c -> match c.value with All | AllOf _ -> true | Expr _ -> false) columns in
  let child_scope =
    match env.scope with
    | (Top_level | From_passthrough) when is_passthrough -> From_passthrough
    | Top_level | From_passthrough | Subquery -> Subquery
  in
  let from_env, p2, resolved_from = eval_nested { env with scope = child_scope } from in
  let env = { from_env with scope = env.scope } in
  let env = { env with query_has_grouping = List.length group > 0 } in
  (* Extract IS NOT NULL predicates from WHERE and HAVING *)
  let not_null_keys_where = extract_not_null_column_keys env where in
  let not_null_keys_having = extract_not_null_column_keys env having in
  let not_null_keys = not_null_keys_where @ not_null_keys_having in
  let projection = make_dynamic_select ~env columns in
  let final_schema = infer_schema ~not_null_keys env projection in
  let final_schema =
    match child_scope with
    | From_passthrough -> final_schema @ From.dynamic_columns resolved_from
    | Top_level | Subquery -> final_schema
  in
  let final_schema' = List.concat_map (function
    | AttrWithSources attr -> [attr]
    | DynamicWithSources (_, l) -> List.map (fun f -> f.Sql.field_attr) l
  ) final_schema in
  (* use schema without aliases here *)
  let p1 = get_params_of_columns env projection in
  let env, p3 = if Dialect.Semantic.is_where_aliases_dialect () then 
    let env = { env with schema = make_unique (Schema.Join.cross env.schema final_schema') } in
    env, get_params_opt { env with set_tyvar_strict = true; } where
  else
    let p3 = get_params_opt { env with set_tyvar_strict = true; 
       (* Some dialects support aliasing *)
      schema = List.filter (fun i -> i.Schema.Source.Attr.sources <> []) env.schema; } where in
    env, p3
  in
  (* ORDER BY, HAVING, GROUP BY allow have column without explicit referring to source if it's specified in SELECT *)
  let env = { env with schema = update_schema_with_aliases env.schema final_schema' } in
  let cardinality =
    match from, where with
    | None, None ->
      `One
    | None, Some _ ->
      `Zero_one
    | Some _, _ when group = [] && exists_grouping projection && not (exists_windowing projection) ->
      `One
      (* TODO: analyse join types to determine if cardinality optimization can be done *)
    | Some ((`Table t, _), []), Some w when matches_at_most_one_row ~env { Sql.table = t; alias = None } w ->
      `Zero_one
    | Some _, _ ->
      `Nat
  in
  let p4 = get_params_l env group in
  let p5 = get_params_opt env having in
  let final_schema, p2 =
    Table_elimination.eliminate ~env ~from:resolved_from ~columns ~where ~group ~having ~order final_schema p2
  in
  (final_schema, p1 @ p2 @ p3 @ p4 @ p5, env, cardinality)

(** @return final schema, params and tables that can be referenced by outside scope *)
and resolve_source env (x, alias) =
  let resolve_schema_with_alias schema = begin match alias with 
    | Some { table_name; column_aliases = Some col_schema } -> 
      let schema = Schema.compound ((List.map (fun attr -> Schema.Source.Attr.{sources=[]; attr;})) col_schema) schema in
      schema, [table_name, Schema.Source.from_schema schema]
    | Some { table_name; column_aliases = None } -> 
      let schema = List.map (fun i -> { i with Schema.Source.Attr.sources = table_name :: i.Schema.Source.Attr.sources }) schema in
      schema, [table_name, Schema.Source.from_schema schema]
    | None -> schema, [] 
  end in
  match x with
  | `Select select ->
    let (s,p,_) = eval_select_full env select in
    let tbl_alias = Option.map (fun { table_name; _ } -> table_name) alias in
    let add_src i = { i with Schema.Source.Attr.sources = option_list tbl_alias @ i.Schema.Source.Attr.sources } in
    let s, dyn = List.partition_map (function
      | AttrWithSources a -> Left (add_src a)
      | DynamicWithSources (dp, cols) -> Right (DynamicWithSources (dp, List.map (fun f -> { f with Sql.field_attr = add_src f.Sql.field_attr }) cols))
    ) s in
    let s, tables = resolve_schema_with_alias s in
    { rsrc_schema = s; rsrc_params = p; rsrc_tables = tables; rsrc_dynamic = dyn; rsrc_physical_table = None }
  | `Nested from ->
    let (env,p,resolved_from) = eval_nested env (Some from) in
    let s = infer_schema ~not_null_keys:[] env [dummy_loc All] in
    if alias <> None then failwith "No alias allowed on nested tables";
    let s = List.map (function 
      | AttrWithSources attr -> attr
       (* TODO: next step optimize it *)
      | DynamicWithSources _ -> failwith "Nested source cannot have dynamic columns"
    ) s in
    { rsrc_schema = s; rsrc_params = p; rsrc_tables = env.tables; rsrc_dynamic = From.dynamic_columns resolved_from; rsrc_physical_table = None }
  | `Table s ->
    let (name,s) = Tables_with_derived.get ~env s in
    let is_cte = List.exists (fun (n, _) -> n = name) env.ctes in
    let alias = Option.map (fun { table_name; _ } -> table_name) alias in
    let sources = (name :: option_list alias) in
    let s3 = List.map (fun attr -> { Schema.Source.Attr.attr; sources }) s  in
    { rsrc_schema = s3; rsrc_params = []; rsrc_tables = List.map (fun name -> name, s) sources; rsrc_dynamic = [];
      rsrc_physical_table = if is_cte then None else Some { Sql.table = name; alias } }
  | `ValueRows { row_constructor_list; row_order; row_limit; } ->
    (* 
      The columns of the table output from VALUES have the implicitly 
      named columns column_0, column_1, column_2, and so on
      https://dev.mysql.com/doc/refman/8.4/en/values.html
    *)
    let exprs_to_cols =
      List.mapi (fun idx e ->
        dummy_loc (Expr (dummy_loc e, Some (Printf.sprintf "column_%d" idx)))
      )
    in
    let dummy_select exprs = { columns = exprs_to_cols exprs; from = None; where = None; group = []; having = None } in
    let (s, p, _) = match row_constructor_list with
      | RowExprList [] -> failwith "Each row of a VALUES clause must have at least one column"
      | RowExprList (exprs :: xs) ->
        let unions = List.map (fun exprs -> `Union, dummy_select exprs ) xs in
        let select = dummy_select exprs in
        let select_complete = { select = select, unions; order=row_order; limit=row_limit; select_row_locking = None } in
        let (s, p, v) = eval_select_full env { select_complete; cte = None } in 
        let s = List.map (function 
          | AttrWithSources attr -> attr 
          | DynamicWithSources _ -> failwith "VALUES cannot have dynamic columns"
        ) s in
        (s, p, v)
      | RowParam { id; types; values_start_pos } ->
        List.map (fun t -> { attr = make_attribute' "" (Source_type.to_infer_type t); Schema.Source.Attr.sources = []}) types, 
          [ TupleList (id, ValueRows { types = List.map Source_type.to_infer_type types; values_start_pos }) ], Stmt.Select `Nat
    in
    let s, tables = resolve_schema_with_alias s in
    { rsrc_schema = s; rsrc_params = p; rsrc_tables = tables; rsrc_dynamic = []; rsrc_physical_table = None }


and eval_select_full env { select_complete; cte } =
  let ctes, p1 = Option.map_default eval_cte ([], []) cte in
  let env = { env with ctes = ctes @ env.ctes } in
  let (s1, p2, env, cardinality) = eval_select ~order:select_complete.order env (fst @@ select_complete.select) in
  eval_compound ~env:{ env with tables = env.tables; } (p1 @ p2, s1, cardinality, select_complete)

and eval_cte { cte_items; is_recursive } = 
  let open Schema.Source in
  List.fold_left begin fun (acc_ctes, acc_vars) cte ->
    let env = { empty_env with ctes = acc_ctes; scope = Subquery } in
    let tbl_name = make_table_name cte.cte_name in
    let a1 = List.map (fun attr -> Attr.{ sources = []; attr }) in
    let s1, p1, _kind =
      if is_recursive then 
      begin
        match cte.stmt with
        | CteInline ({ select = select, other; _ } as stmt_) ->
          let other = other |> List.map begin 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
          end
          in
          let stmt = { stmt_ with select = select, other } in
          let s1, p1, env, cardinality = eval_select ~order:[] env (fst stmt.select) in
          let s1' = List.map (function 
            | AttrWithSources attr -> attr
            (* TODO: next step is to support it for CTEs *)
            | DynamicWithSources _ -> failwith "Recursive CTEs cannot have dynamic columns"
          ) s1 in
          (* UNIONed fields access by alias to itself cte *)
          let s2 = Schema.compound (Option.map_default a1 s1' cte.cols) s1' in
          let a2 = from_schema s2 in
          eval_compound ~env:{ env with ctes = (tbl_name, a2) :: env.ctes } (p1, s1, cardinality, stmt) 
        | CteSharedQuery _ -> failwith "Recursive CTEs with shared query currently are not supported"
      end    
      else (
        match cte.stmt with
        | CteInline stmt ->
          let s1, p1, env, cardinality = eval_select ~order:[] env (fst stmt.select) in
          eval_compound ~env:{ env with tables = env.tables } (p1, s1, cardinality, stmt)
        | CteSharedQuery shared_query_name ->
          let (_, stmt) = Shared_queries.get shared_query_name.value in
          let s1, p1, kind = eval_select_full env stmt in
          s1, [SharedVarsGroup (p1, shared_query_name)], kind
      )
    in
    let s1 = List.map (function 
      | AttrWithSources attr -> attr
        (* TODO: next step is to support it for CTEs *)
      | DynamicWithSources _ -> failwith "Recursive CTEs cannot have dynamic columns"
    ) s1 in
    let s2 = Schema.compound (Option.map_default a1 s1 cte.cols) s1 in
    (tbl_name, from_schema s2) :: acc_ctes, acc_vars @ p1 end
  ([], []) cte_items  

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 ~order:[] env) other) in
  let cardinality = if other = [] then cardinality else `Nat in
  (* ignoring tables in compound statements - they cannot be used in ORDER BY *)
  let final_schema = 
    if other = [] then s1
    else (
      (* TODO: next step is to support it for UNIONS (but if it's possible to control it) *)
      let unwrap_attr = function 
        | AttrWithSources attr -> attr
        | DynamicWithSources _ -> failwith "Union/Except/Intersect doesn't support dynamic columns"
      in
      let s1' = List.map unwrap_attr s1 in
      let s2l' = List.map (List.map unwrap_attr) s2l in
      List.map (fun x -> AttrWithSources x) @@ List.fold_left Schema.compound s1' s2l'
    )
  in
  let p3 = 
    let schema = List.concat_map (function 
      | AttrWithSources attr -> [attr]
      | DynamicWithSources (_, a) -> List.map (fun f -> f.Sql.field_attr) a
    ) final_schema in
    params_of_order order schema env in
  let (p4,limit1) = match limit with Some (p,x) -> List.map (fun p -> 
    Single (make_param ~id:p.id ~typ:(Source_type.to_infer_type p.typ), Meta.empty())) 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  

let update_tables ~env sources ss w =
  let schema = Schema.cross_all @@ List.map (fun src -> src.rsrc_schema) sources in
  let p0 = List.flatten @@ List.map (fun src -> src.rsrc_params) sources in
  let tables = List.flatten @@ List.map (fun src -> src.rsrc_tables) sources in (* TODO assert equal duplicates if not unique *)
  let env = { env with tables; schema; } 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

let annotate_select select attrs =
  let (select1,compound) = select.select in
  let apply_to_columns cols attrs =
    let rec loop acc cols attrs =
      match cols, attrs with
      | [], [] -> List.rev acc
      | ({ value = (All | AllOf _); _ }) :: _, _ -> failwith "Asterisk not supported"
      | { value = Expr (loc, name); pos = col_pos } :: cols, a :: attrs ->
        let e = merge_meta_into_params ~shallow:false a.meta loc.value in
        let t = a.domain in
        loop ({ value = Expr ({ loc with value = Fun { fn_name = "insert_select"; kind = (F (Typ t, [Typ t])); parameters = [e]; is_over_clause = false} }, name); pos = col_pos } :: acc) cols attrs
      | _, [] | [], _ -> failwith "Select cardinality doesn't match Insert"
    in
    loop [] cols attrs
  in
  let select1' = { select1 with columns = apply_to_columns select1.columns attrs } in
  let compound' = List.map (fun (op, sel) -> (op, { sel with columns = apply_to_columns sel.columns attrs })) compound in
  { select with select = (select1', compound') }

let resolve_on_conflict_clause ~env tn' = Option.map_default (function
  | {value = On_conflict { action; attrs; }; _ } -> 
    let names = List.map (fun attr -> attr.cname) attrs in
    let composite_primary_key = Constraint.make_composite_primary names in
    let composite_unique = Constraint.make_composite_unique names in
    List.iter (fun col -> 
      let resolved = resolve_column ~env col in
      if (Constraints.disjoint (Constraints.of_list [
        Unique; PrimaryKey;
        composite_primary_key;
        composite_unique
      ]) resolved.attr.extra ) then
        fail "Schema Error: ON CONFLICT clause (%s) does not match the PRIMARY KEY or UNIQUE constraint for column: %s"
          (names |> String.concat ", ")
          (show_col_name col)
    ) attrs;
    begin match action with
    | Do_nothing -> []
    | Do_update values -> 
        let ss = List.map (function
          (*
            The SET and WHERE clauses in ON CONFLICT DO UPDATE have access 
            to the existing row using the table's name (or an alias),
            and to rows proposed for insertion using the special excluded table.
            From our perspective, it is the same as accessing the table into which we write.
          *)
         | col, RegularExpr (Column { collated = { cname ; tname = Some { tn = "excluded"; db }; }; collation }) -> 
          col, RegularExpr(Column { collated = { cname; tname = Some { tn = tn'; db }; }; collation })
         | e -> e
        ) values in
        ss
    end
  | { value = On_duplicate { assignments; }; _ } -> assignments
) []

let with_constraints attrs constraints : Schema.t =
  let constraints_table : (string, Constraints.t) Hashtbl.t = Hashtbl.create (List.length attrs) in
  List.iter (fun attr ->
    Hashtbl.replace constraints_table attr.name attr.extra
  ) attrs;
  List.iter (fun constr ->
    match constr with
    | `Primary [] -> fail "Schema Error: PRIMARY KEY must have at least one column"
    | `Unique (_, []) -> fail "Schema Error: UNIQUE constraint must have at least one column"
    | `Primary [ col_name ] -> begin
      match Hashtbl.find_opt constraints_table col_name with
      | None -> fail "Schema Error: no such column: %s" col_name
      | Some constraints ->
        let new_constraints = Constraints.add PrimaryKey constraints in
        Hashtbl.replace constraints_table col_name new_constraints
      end
    | `Unique (_, [ col_name ]) -> begin
      match Hashtbl.find_opt constraints_table col_name with
      | None -> fail "Schema Error: no such column: %s" col_name
      | Some constraints ->
        let new_constraints = Constraints.add Unique constraints in
        Hashtbl.replace constraints_table col_name new_constraints
      end
    | `Primary cols -> begin
      List.iter (fun col ->
        match Hashtbl.find_opt constraints_table col with
        | None -> fail "Schema Error: no such column: %s" col
        | Some constraints ->
          let new_constraints = Constraints.add (Constraint.make_composite_primary cols) constraints in
          Hashtbl.replace constraints_table col new_constraints
      ) cols
    end
    | `Unique (_, cols) -> begin
      List.iter (fun col ->
        match Hashtbl.find_opt constraints_table col with
        | None -> fail "Schema Error: no such column: %s" col
        | Some constraints ->
          let new_constraints = Constraints.add (Constraint.make_composite_unique cols) constraints in
          Hashtbl.replace constraints_table col new_constraints
      ) cols
    end
    | `Ignore -> ()
  ) constraints;
  List.map (fun attr ->
    match Hashtbl.find_opt constraints_table attr.name with
    | Some constraints -> { attr with extra = constraints }
    | None -> attr
  ) attrs


let rec eval (stmt:Sql.stmt) =
  let open Stmt in
  let open Schema.Source in
  let open Attr in
  match stmt with
  | Create (name, Schema { schema; constraints; indexes }) ->
      let attrs = List.map Alter_action_attr.to_attr schema in
      let attrs = with_constraints attrs constraints in
      let columns = List.map2 (fun (col : Alter_action_attr.t) attr ->
        {
          Tables.attr;
          source_kind = Option.map (fun k -> k.value) col.kind;
          default_sql = Alter_action_attr.default_sql col;
        }
      ) schema attrs in
      Tables.add_columns (name, columns);
      Tables.add_inline_indexes name ~indexes ~constraints;
      ([],[],Create name)
  | Create (name, Select { value=select; _ }) ->
      let (schema,params,_) = eval_select_full empty_env select in
      let schema = List.map (function 
        | AttrWithSources attr -> attr 
        | DynamicWithSources _ -> failwith "CREATE TABLE AS SELECT cannot have dynamic columns"
      ) schema in
      Tables.add (name, from_schema schema);
      ([],params,Create name)
  | Alter (name,actions) ->
      List.iter (function
      | `Add (col,pos) ->
        let source_kind = Option.map (fun k -> k.value) col.Alter_action_attr.kind in
        let default_sql = Alter_action_attr.default_sql col in
        Tables.alter_add name ~col:{ attr = Alter_action_attr.to_attr col; source_kind; default_sql } ~pos
      | `Drop col ->
        Tables.alter_drop name ~col
      | `Change (oldcol,col,pos) ->
        let source_kind = Option.map (fun k -> k.value) col.Alter_action_attr.kind in
        let default_sql = Alter_action_attr.default_sql col in
        Tables.alter_change name ~oldcol ~col:{ attr = Alter_action_attr.to_attr col; source_kind; default_sql } ~pos
      | `RenameColumn (oldcol,newcol) ->
        Tables.rename_column name ~old_name:oldcol ~new_name:newcol
      | `RenameTable new_name ->
        Tables.rename name new_name
      | `DropPrimaryKey ->
        Tables.drop_primary_key name
      | `AddPrimaryKey cols ->
        Tables.add_primary_key name ~cols
      | `AlterColumnPG (col_name, change) ->
        Tables.alter_column_pg name ~col_name change.value
      | `AddIndex { add_idx_name = Some index_name; add_idx_kind = kind; add_idx_cols = cols } ->
        Tables.index_add name ~index_name ~kind ~cols
      | `AddIndex { add_idx_name = None; add_idx_kind = kind; add_idx_cols = cols } ->
        Tables.index_add_auto name ~kind ~cols
      | `DropIndex index_name ->
        Tables.index_drop name ~index_name
      | `RenameIndex (old_name, new_name) ->
        Tables.index_rename name ~old_name ~new_name
      | `AddConstraint _ | `DropConstraint _ -> ()
      | `TtlOptions (opts, _) ->
        let expr, enabled =
          List.fold_left (fun (expr, enabled) -> function
            | `TtlSet (col, n, unit) -> Some (col, n, String.uppercase_ascii unit), enabled
            | `TtlEnable v -> expr, Some (String.uppercase_ascii v <> "OFF"))
            (None, None) opts
        in
        let prev = Tables.get_ttl name in
        let ttl_enabled =
          Option.default (Option.map_default (fun (t : Tables.table_ttl) -> t.ttl_enabled) true prev) enabled
        in
        Tables.set_ttl name @@
          Option.map_default
            (fun (ttl_col, ttl_n, ttl_unit) -> Some { Tables.ttl_col; ttl_n; ttl_unit; ttl_enabled })
            (Option.map (fun (t : Tables.table_ttl) -> { t with ttl_enabled }) prev)
            expr
      | `RemoveTtl _ -> Tables.set_ttl name None
      | `Default_or_convert_to (cs, collation) ->
        let old = Tables.get_charset name in
        let collation =
          match Option.map (fun c -> c.value) collation with
          | Some _ as c -> c
          | None -> Option.map_default (fun o -> o.Tables.collation) None old
        in
        Tables.set_charset name { charset = cs; collation }) actions;
      ([],[],Alter [name])
  | Rename l ->
    List.iter (fun (o,n) -> Tables.rename o n) l;
    ([], [], Alter (List.map fst l)) (* to have sensible target for gen_xml *)
  | Drop name ->
      Tables.drop name;
      ([],[],Drop name)
  | CreateIndex { ci_name; ci_table; ci_cols; ci_kind } ->
      let cols = List.map (fun x -> x.collated) ci_cols in
      Sql.Schema.project cols (Tables.get_schema ci_table) |> ignore;
      Tables.index_add ci_table ~index_name:ci_name ~kind:ci_kind ~cols;
      [],[],CreateIndex ci_name
  | Insert { target=table; action=`Values (names, values); on_conflict_clause; _ } ->
    let expect = values_or_all table names in
    let t = Tables.get_schema table in
    let schema = List.map (fun attr -> { sources=[table]; attr }) t in
    let env = { empty_env with tables = [Tables.get table]; schema; } in
    begin match values with
    | None -> 
      [], [], Insert(Some (Values, expect), table)
    | Some values ->
      let vl = List.map List.length values in
      let cl = List.length expect in
      if List.exists (fun n -> n <> cl) vl then
        fail "Expecting %u expressions in every VALUES tuple" cl;
      (* pair up columns with inserted values *)
      let assigns = values |> List.map (fun tuple ->
        List.combine 
        (List.map (fun a -> {cname=a.name; tname=None}) expect)
        tuple
      ) in
      let resolved = List.concat_map (fun l -> 
        let resolved = resolve_column_assignments ~env l in 
        List.map2 (fun e (c, _) -> 
          let (res, t) = resolve_types env e in 
          let params = get_params_of_res_expr env res in
          c, params, get_or_failwith t
        ) resolved l
      ) assigns in
      (*
        DDL:
        -- [sqlgg] non_nullifiable=true
        col INT NULL
      
        INSERT with multiple VALUES:
        INSERT INTO t (col) VALUES 
          (42),     -- col: Int (strict)
          (NULL),   -- col: Int? (nullable) 
          (@param); -- col: Int? (inferred as nullable)
      
        Flow:
        DDL column type: Int?          (nullable from schema)
          ↓
        VALUES row 1:    Int           (strict literal)
        ....
          ↓
        Aggregated type: Int?          (nullable wins)
          ↓
        VALUES(col) type: Int?         (overall nullable)

        - ON DUPLICATE KEY UPDATE - this is part of UPDATE and should be checked then for non_nullifiable
        this is why we build env.insert_resolved_types
      *)
      List.iter (fun (c, _, t) -> 
        match Hashtbl.find_opt env.insert_resolved_types c.cname with
        | None -> 
          Hashtbl.add env.insert_resolved_types c.cname t
        | Some t0 -> 
          Hashtbl.replace env.insert_resolved_types c.cname 
          { t with Type.nullability = Type.common_nullability [t; t0] }
      ) resolved;
      let p1 = List.concat_map (fun (_c, p, _t) -> p) resolved in
      let conflict_assigns = resolve_on_conflict_clause ~env table.tn on_conflict_clause in
      let params2 = params_of_assigns { env with is_update = true; } conflict_assigns in
      [], p1 @ params2, Insert (None, table)
    end
  | Insert { target=table; action=`Param (names, param_id); on_conflict_clause; _ } ->
    let schema = List.map (fun attr -> { Schema.Source.Attr.sources=[table]; attr }) (Tables.get_schema table) in
    let env = { empty_env with tables = [Tables.get table]; schema; } in
    let conflict_assigns = resolve_on_conflict_clause ~env table.tn on_conflict_clause in
    let expect = values_or_all table names in
    List.iter (fun a -> Hashtbl.add env.insert_resolved_types a.attr.name a.attr.domain ) schema;
    let params2 = params_of_assigns { env with is_update = true } conflict_assigns in
    let params = [ TupleList (param_id, Insertion expect) ] in
    [], params @ params2, Insert (None, table)
  | Insert { target=table; action=`Select (names, select); on_conflict_clause; _ } ->
    let expect = values_or_all table names in
    let env = { empty_env with tables = [Tables.get table]; 
      schema = List.map (fun attr -> { sources=[table]; attr }) (Tables.get_schema table); 
    } in
    let select_complete = annotate_select select.select_complete expect in
    let select = { select with select_complete } in
    let (schema,params,_) = eval_select_full env select in
    let schema = List.map (function 
      | AttrWithSources attr -> attr 
      | DynamicWithSources _ -> failwith "INSERT ... SELECT cannot have dynamic columns"
    ) schema in
    ignore (Schema.compound
      ((List.map (fun attr -> {sources=[]; attr;})) expect)
      (List.map (fun {attr; _} -> {sources=[]; attr}) schema)); (* test equal types once more (not really needed) *)
    let conflict_assigns = resolve_on_conflict_clause ~env table.tn on_conflict_clause in
    List.iter2 (fun a1 a2 -> Hashtbl.add env.insert_resolved_types a2.name a1.attr.domain ) schema expect;
    let params2 = params_of_assigns { env with is_update = true } conflict_assigns in
    [], params @ params2, Insert (None,table)
  | Insert { target=table; action=`Set ss; on_conflict_clause; _ } ->
    let env = { empty_env with tables = [Tables.get table]; 
      schema = List.map (fun attr -> { sources=[table]; attr }) (Tables.get_schema table);
    } in
    let (params,inferred) = match ss with
    | None -> [], Some (Assign, Tables.get_schema table)
    | Some ss -> params_of_assigns env ss, None
    in
    let conflict_assigns = resolve_on_conflict_clause ~env table.tn on_conflict_clause in
    let params2 = params_of_assigns { env with is_update = true } conflict_assigns in
    [], params @ params2, Insert (inferred,table)
  | Delete (table, where) ->
    let t = Tables.get table in
    let p = get_params_opt { empty_env with tables=[t]; 
      schema=List.map (fun attr -> { Schema.Source.Attr.sources=[t |> fst]; attr }) (t |> snd); 
      set_tyvar_strict = true 
    } where in
    [], p, Delete [table]
  | DeleteMulti (targets, tables, where) ->
    (* use dummy columns to verify targets match the provided tables  *)
    let select = ({ columns = [dummy_loc All]; from = Some tables; where; group = []; having = None }, []) in
    let select_complete = { select; order = []; limit = None; select_row_locking = None } in
    let _attrs, params, _ = eval_select_full empty_env {select_complete; cte=None } in
    [], params, Delete targets
  | Set (vars, stmt) ->
    let p =
      vars |> List.map (fun (_k,e) ->
        match e with
        | Column _ -> [] (* this is not column but some db-specific identifier *)
        | _ -> get_params_of_res_expr empty_env (ensure_res_expr e)) |> List.concat
    in
    begin match stmt with
    | None -> [], p, Other
    | Some stmt -> let (schema,p2,kind) = eval stmt in (schema, p @ p2, kind)
    end
  | Update (table,ss,w,o,lim) ->
    let f, s = Tables.get table in
    let env = { empty_env with is_update = true } in
    let r = List.map (fun attr -> {Schema.Source.Attr.attr; sources=[f] }) s in
    let params = update_tables ~env [{ rsrc_schema = r; rsrc_params = []; rsrc_tables = [(f, s)]; rsrc_dynamic = [];
      rsrc_physical_table = Some { Sql.table = f; alias = None } }] ss w in
    let env = { env with schema = update_schema_with_aliases [] r; is_update = true } in
    let p3 = params_of_order o [] { env with tables = [(f, s)] } in
    let lim = List.map (fun p -> make_param ~id:p.id ~typ:(Source_type.to_infer_type p.typ)) lim in
    [], params @ p3 @ (List.map (fun p -> Single (p, Meta.empty())) lim), Update (Some table)
  | UpdateMulti (tables,ss,w,o,lim) ->
    let env = { empty_env with is_update = true } in
    let sources = List.map (fun src -> resolve_source { env with scope = Subquery } ((`Nested src), None)) tables in
    let tables = List.map (fun src -> src.rsrc_tables) sources |> List.flatten in
    let params = update_tables ~env sources ss w in
    let p3 = params_of_order o [] { env with schema = Schema.cross_all @@ List.map (fun src -> src.rsrc_schema) sources; tables } in
    let lim = List.map (fun p -> make_param ~id:p.id ~typ:(Source_type.to_infer_type p.typ)) lim in
    [], params @ p3 @ (List.map (fun p -> Single (p, Meta.empty())) lim), Update None
  | Select select -> 
    let (schema, a, b) = eval_select_full empty_env select in
    List.map drop_sources schema, a, b
  | CreateRoutine (name,_,_) ->
    [], [], CreateRoutine name
  | CreateType (name, TypeEnum ctors) ->
     User_types.add name (Sql.Type.make_enum_kind ctors);
     ([], [], CreateType name)
  | DropType (name, if_exists) ->
     User_types.drop ~if_exists name;
     ([], [], DropType name)

type var_shape =
  | Shape_param
  | Shape_in_param
  | Shape_tuple of string option
  | Shape_choice_in of { param : string option; kind : in_or_not_in; vars : var_shape list }
  | Shape_opt_choice of string option * var_shape list
  | Shape_shared_group of string * var_shape list
  | Shape_choice of string option * ctor_shape list
  | Shape_dyn_select of string option * ctor_shape list
  | Shape_dyn_join of string option
and ctor_shape =
  | Shape_simple of string option * var_shape list
  | Shape_verbatim of string

let rec var_shape = function
  | Single _ -> Shape_param
  | SingleIn _ -> Shape_in_param
  | TupleList (id, _) -> Shape_tuple id.value
  | ChoiceIn { param; kind; vars } -> Shape_choice_in { param = param.value; kind; vars = List.map var_shape vars }
  | OptionActionChoice (id, vars, _, _) -> Shape_opt_choice (id.value, List.map var_shape vars)
  | SharedVarsGroup (vars, id) -> Shape_shared_group (id.value, List.map var_shape vars)
  | Choice (id, cs) -> Shape_choice (id.value, List.map ctor_shape cs)
  | DynamicSelect (id, cs) -> Shape_dyn_select (id.value, List.map ctor_shape cs)
  | DynamicSelectJoin { pid; _ } -> Shape_dyn_join pid.value
and ctor_shape = function
  | Simple (p, args) -> Shape_simple (p.value, List.map var_shape (Option.default [] args))
  | Verbatim (n, _) -> Shape_verbatim n

module Var_unifier : sig
  type t
  val create : unit -> t
  val note : t -> string -> Type.t -> unit
  val alias : t -> string -> string -> unit
  val typ : t -> string -> default:Type.t -> Type.t
end = struct
  type node = { name : string; mutable state : state }
  and state = Root of Type.t option | Link of node

  type t = (string, node) Hashtbl.t

  let create () = Hashtbl.create 10

  let node t name =
    match Hashtbl.find_opt t name with
    | Some n -> n
    | None -> let n = { name; state = Root None } in Hashtbl.add t name n; n

  (* one node per variable, so nodes are compared physically *)
  let rec root n =
    match n.state with
    | Root typ -> n, typ
    | Link p -> let (r, _) as res = root p in (* relink directly to root so next lookups are one step *) if r != p then n.state <- Link r; res

  let unify name t1 t2 =
    match Type.common_type t1 t2 with
    | Some x ->
      if !Config.debug then eprintfn "unify var %s %s %s => %s" name (Type.show t1) (Type.show t2) (Type.show x);
      x
    | None -> fail "incompatible types for parameter %S : %s and %s" name (Type.show t1) (Type.show t2)

  let note t name typ =
    let (r, existing) = root (node t name) in
    r.state <- Root (Some (Option.map_default (unify name typ) typ existing))

  let alias t n1 n2 =
    let (r1, _) = root (node t n1) in
    let (r2, t2) = root (node t n2) in
    if r1 != r2 then begin (* physically same root = already aliased *)
      Option.may (note t r1.name) t2;
      r2.state <- Link r1
    end

  let typ t name ~default =
    Stdlib.Option.bind (Hashtbl.find_opt t name) (snd $ root)
    |> Option.default default
end

(* FIXME unify each choice separately *)
let unify_params l =
  if !Config.debug then l |> List.iter (fun p -> eprintfn "var %s" (show_var p));
  let unifier = Var_unifier.create () in
  let choices = Hashtbl.create 10 in
  let rec bound_names vars =
    vars |> List.concat_map (function
      | Single (p, _) | SingleIn (p, _) -> [p.id.value]
      | v -> bound_names (sub_vars v))
  in
  let register p signature =
    match p.value with
    | None -> () (* anonymous ie non-shared *)
    | Some n ->
    match Hashtbl.find_opt choices n, signature with
    | None, _ -> Hashtbl.add choices n signature
    | Some (`Branches (s1, names1)), `Branches (s2, names2) when s1 = s2 ->
      List.iter2 (fun n1 n2 -> match n1, n2 with Some n1, Some n2 -> Var_unifier.alias unifier n1 n2 | _ -> ()) names1 names2
    | Some (`Branches _), `Branches _ -> failed ~at:p.pos "choice %s is used several times with different branches" n
    | Some `Dynamic, `Dynamic -> failed ~at:p.pos "dynamic select %s occurs several times in one statement (not supported)" n
    | Some `Dynamic, `Branches _ | Some (`Branches _), `Dynamic ->
      if n = dynamic_col_param_name then
        failed ~at:p.pos "dynamic_select reserves the name %s for the column picker, rename choice %s" n n
      else
        failed ~at:p.pos "parameter %s is ambiguous : used as both choice and dynamic select" n
  in
  let rec collect var =
    begin match var with
    | Single ({ id; typ; _ }, _) | SingleIn ({ id; typ; _ }, _) ->
      Option.may (fun name -> Var_unifier.note unifier name typ) id.value
    | Choice (p, ctors) ->
      register p (`Branches (List.map ctor_shape ctors, bound_names (List.concat_map ctor_vars ctors)))
    | DynamicSelect (p, _) -> register p `Dynamic
    | TupleList _ | ChoiceIn _ | OptionActionChoice _ | SharedVarsGroup _ | DynamicSelectJoin _ -> ()
    end;
    List.iter collect (sub_vars var)
  in
  (* if no other clues - input parameters are strict *)
  let final { id; typ; _ } =
    let typ = Option.map_default (Var_unifier.typ unifier ~default:typ) typ id.value in
    make_param ~id ~typ:(Type.undepend typ Strict)
  in
  let rec rewrite = function
    | Single (p, m) -> Single (final p, m)
    | SingleIn (p, m) -> SingleIn (final p, m)
    | v -> map_sub_vars (List.map rewrite) v
  in
  List.iter collect l;
  List.map rewrite l

let is_alpha = function
| 'a'..'z' -> true
| 'A'..'Z' -> true
| _ -> false

let common_prefix = function
| [] -> 0
| x::_ as l ->
  let rec loop i =
    if String.length x <= i then i
    else
      if List.for_all (fun s -> i < String.length s && s.[i] = x.[i]) l then
        loop (i+1)
      else
        i
  in
  let i = loop 0 in
  (* do not allow empty names or starting not with alpha *)
  if List.exists (fun s -> i = String.length s || not (is_alpha s.[i])) l then 0 else i

(* fill inferred sql for VALUES or SET *)
let complete_sql kind sql =
  match kind with
  | Stmt.Insert (Some (kind,schema), _) ->
    let (pre,each,post) = match kind with
    | Values -> "(", (fun _ -> ""), ")"
    | Assign -> "", (fun name -> name ^" = "), ""
    in
    let module B = Buffer in
    let b = B.create 100 in
    B.add_string b sql;
    B.add_string b " ";
    B.add_string b pre;
    let params = ref [] in
    let first = common_prefix @@ List.map (fun attr -> attr.Sql.name) schema in
    schema |> List.iter (fun attr ->
      if !params <> [] then B.add_string b ",";
      let attr_ref_prefix = each attr.Sql.name in
      let attr_name = String.slice ~first attr.Sql.name in
      let attr_ref = "@" ^ attr_name in
      let pos_start = B.length b + String.length attr_ref_prefix in
      let pos_end = pos_start + String.length attr_ref in
      (* autoincrement is special - nullable on insert, strict otherwise *)
      let typ = if Constraints.mem Autoincrement attr.extra then Sql.Type.nullable attr.domain.t else attr.domain in
      let param = Single (make_param ~id:{value=Some attr_name; pos=(pos_start,pos_end)} ~typ, Meta.empty()) in
      B.add_string b attr_ref_prefix;
      B.add_string b attr_ref;
      tuck params param;
    );
    B.add_string b post;
    (B.contents b, List.rev !params)
  | _ -> (sql,[])

let eval_parsed sql ({ Parser.statement; dialect_features } : Parser.parse_result) =
  let (schema,p1,kind) = eval statement in
  let (sql,p2) = complete_sql kind sql in
  (sql, schema, unify_params (p1 @ p2), kind, dialect_features)

let parse sql =
  eval_parsed sql (Parser.parse_stmt sql)

let eval_select select_full =
  let (schema, p1, kind) = eval @@ Select select_full in
  (schema, unify_params p1, kind)