path: root/lib/analysers/query.ml

open StdLabels
module Expression = ImportExpression
module Q = Expression.Query
module Syntax = ImportConf.Syntax
module Table = ImportConf.Table
module Path = ImportConf.Path

let truncate buffer n = Buffer.truncate buffer (Buffer.length buffer - n)

(** The module allow to create fragment in the query which keep together the
    binderd parameters and the text of the query.contents.

    This is used a lot in order to create the CTE, where you need the create
    fragment used both in the main request and partially in the CTE itself.

    The content is mutable and all the functions are returning [unit]. *)
module Chunk = struct
  type t = {
    b : Buffer.t;
    parameters : ImportCSV.DataType.t Queue.t;
  }

  let create : unit -> t =
   fun () -> { b = Buffer.create 16; parameters = Queue.create () }

  let create' : Buffer.t -> ImportCSV.DataType.t Queue.t -> t =
   fun b parameters -> { b; parameters }

  (* Append the element from [tail] at the end of [head]

     Tail is destroyed during the operation.
  *)
  let append : head:t -> tail:t -> unit =
   fun ~head ~tail ->
    match Buffer.length tail.b with
    | 0 -> ()
    | _ ->
        Buffer.add_buffer head.b tail.b;
        Queue.transfer tail.parameters head.parameters;
        ()

  (** Add a litteral string in the sequence *)
  let add_string : t -> string -> unit = fun t v -> Buffer.add_string t.b v

  let copy : t -> t =
   fun t ->
    let b = Buffer.create 16 and parameters = Queue.copy t.parameters in
    Buffer.add_buffer b t.b;
    { b; parameters }

  let add_parameters : t -> ImportCSV.DataType.t Seq.t -> unit =
   fun t p -> Queue.add_seq t.parameters p
end

let prepare_key : f:(Format.formatter -> unit) -> Format.formatter -> unit =
 fun ~f formatter -> Format.fprintf formatter "rtrim(upper(%t))" f

(* Collect all the tables pointed by the expression. *)
let pointed_tables : Syntax.t -> 'a Expression.T.t -> (Table.t * string) list =
 fun conf expression ->
  Expression.T.fold_values expression ~init:[] ~f:(fun acc path ->
      let table = ImportConf.get_table_for_name conf path.Path.alias in
      let table_name = Table.name table in
      (table, table_name) :: acc)
  |> List.sort_uniq ~cmp:Stdlib.compare

(** Represent a column in a safe way in a query *)
let print_column : Table.t -> string -> string =
 fun table column ->
  String.concat ~sep:"" [ "'"; table.Table.name; "'.'"; column; "'" ]

let create_table : Dependency.t -> string =
 fun mapping ->
  let b = Buffer.create 64 in

  Buffer.add_string b "CREATE TABLE '";
  Buffer.add_string b (Table.name (Dependency.table mapping));
  Buffer.add_string b "' (id INTEGER PRIMARY KEY";

  List.iter (Dependency.keys mapping) ~f:(fun { Dependency.name; _ } ->
      Buffer.add_string b ",'key_";
      Buffer.add_string b name;
      Buffer.add_string b "'");

  ImportContainers.IntSet.iter (Dependency.columns mapping) ~f:(fun i ->
      Buffer.add_string b ",'col_";
      Buffer.add_string b (string_of_int i);
      Buffer.add_string b "'");
  Buffer.add_string b ")";

  Buffer.contents b

let show_path : conf:Syntax.t -> Format.formatter -> Path.t -> unit =
 fun ~conf buffer { alias; column } ->
  let table = ImportConf.get_table_for_name conf alias in
  let table_name = table.Table.name in
  Format.fprintf buffer "'%s'.col_%d" table_name column

(** Extract the informations from the dependancies. We get two informations here
    :

    - the join query in order to load the data from the external column
    - the column corresponding to the key in order to identify the missing links
      later. *)
let query_of_external :
    conf:Syntax.t -> join_buffer:Chunk.t -> Syntax.Extern.t -> unit =
 fun ~conf ~join_buffer external_ ->
  let extern_table = Table.name external_.target in

  let formatter = Format.formatter_of_buffer join_buffer.b in
  Format.fprintf formatter "\nLEFT JOIN '%s' AS '%s' ON %t = %s" extern_table
    external_.target.name
    (prepare_key ~f:(fun f ->
         let q =
           Q.query_of_expression Q.BindParam f (show_path ~conf)
             external_.intern_key
         in

         Chunk.add_parameters join_buffer (Queue.to_seq q)))
    (print_column external_.Syntax.Extern.target
       ("key_" ^ external_.Syntax.Extern.target.name));

  Format.pp_print_flush formatter ()

(** Create the from part of the query, adding all the reuired externals.

    SQLite is able to optimize the query and do not load the table not used in
    the select clause. *)
let create_from_chunck : Syntax.t -> Chunk.t -> unit =
 fun conf c ->
  Chunk.add_string c "\nFROM '";
  Chunk.add_string c (Table.name conf.source);
  Chunk.add_string c "' AS '";
  Chunk.add_string c conf.source.name;
  Chunk.add_string c "'";

  (* Add the externals in the query *)
  List.iter conf.externals ~f:(query_of_external ~conf ~join_buffer:c)

(** Build a CTE query in order to use any group function inside the query.
    Return the binded parameters used in the expression. The buffer given in
    argument is also modified during the construction.

    If filters is not None, the clauses are added to the CTE. *)
let build_cte :
    Syntax.t ->
    expression:'a Expression.T.t ->
    filters:Chunk.t option ->
    Chunk.t =
 fun conf ~expression ~filters ->
  (* The binded parameters queue will be used later in the full query *)
  let cte_chunk = Chunk.create () in

  Chunk.add_string cte_chunk "WITH cte AS (SELECT ";
  Chunk.add_string cte_chunk conf.source.name;
  Chunk.add_string cte_chunk ".id, ";

  let formatter = Format.formatter_of_buffer cte_chunk.b in

  let p =
    Q.query_of_expression Q.BindParam formatter (show_path ~conf) expression
  in
  Format.pp_print_flush formatter ();
  Chunk.add_parameters cte_chunk (Queue.to_seq p);
  (* The name is hardcoded here, and used in [Expression.Filters.window] *)
  Chunk.add_string cte_chunk " AS group0";

  let () = create_from_chunck conf cte_chunk in
  let () =
    match filters with
    | None -> ()
    | Some filters_chunk ->
        Chunk.append ~head:cte_chunk ~tail:(Chunk.copy filters_chunk)
  in

  Chunk.add_string cte_chunk ")\n";
  cte_chunk

type filter_evaluation = {
  content : Buffer.t;
  parameters : ImportCSV.DataType.t Seq.t;
  cte : (string * Chunk.t) option;
}
(** Build the filters to apply in the query. We make the difference here between
    the predicates to apply directly in the query, and the filters associated
    with a group, which are required to be transformed into a CTE in SQL, and
    are evaluated before. *)

(** Evaluate the filters on the query *)
let eval_filters : Syntax.t -> filter_evaluation =
 fun conf ->
  match conf.filters with
  | [] ->
      let empty_buffer = Buffer.create 0 in
      { content = empty_buffer; parameters = Seq.empty; cte = None }
  | filters -> (
      (* Create a new queue in order to accumulate all the parameters to bind.
         This filter will be given to both the CTE if any, or reused in the
         main query when there is no CTE.
      *)
      let chunk_filters = Chunk.create () in
      Chunk.add_string chunk_filters "\nWHERE ";

      let group = Chunk.create () in

      let with_cte, with_exr =
        List.fold_left filters ~init:(None, false)
          ~f:(fun (with_cte, with_exr) column ->
            (* The function will return an option in second position which is
               None when no Group function where found, and Some Expression
               otherwise *)
            let b = Buffer.create 16 in

            let formatter = Format.formatter_of_buffer b in
            let queue, group_found =
              Expression.Filters.query_of_expression Q.BindParam formatter
                (show_path ~conf) column
            in
            Format.pp_print_flush formatter ();
            let clause = Chunk.create' b queue in

            match (group_found, with_cte) with
            | None, _ ->
                Chunk.append ~head:chunk_filters ~tail:clause;
                Chunk.add_string chunk_filters "\nAND ";
                (with_cte, true)
            | (Some _ as group'), None ->
                (* We have a group here, we do not add it into the
                   filter_buffer right now.

                   This can occur only once, the second one will raise
                   an error. *)
                Chunk.append ~head:group ~tail:clause;
                (group', with_exr)
            | Some _, Some _ -> raise ImportErrors.MisplacedWindow)
      in

      match with_cte with
      | None ->
          let content = chunk_filters.b in
          truncate content 5;
          {
            (* There is no group clause in the query *)
            content;
            parameters = Queue.to_seq chunk_filters.parameters;
            cte = None;
          }
      | Some expression ->
          let filters =
            if with_exr then (
              (* If we have additionnals filters from the group clause, we
                 have to report them in the CTE instead of the main query. *)
              let c' = Chunk.copy chunk_filters in
              truncate c'.b 5;
              Some c')
            else None
          in

          (* Create the common expression table *)
          let cte_parameters = build_cte conf ~expression ~filters in
          Chunk.append ~head:chunk_filters ~tail:group;

          {
            content = chunk_filters.b;
            parameters = Queue.to_seq chunk_filters.parameters;
            (* The name is hardcoded here, and used in
               [Expression.Filters.window] *)
            cte = Some ("cte", cte_parameters);
          })

type query = {
  q : string;
  parameters : ImportCSV.DataType.t Seq.t;
}

(** Build the query and return also the mapping in order to identify each
    external links between files.

    The select query will name each column with an alias, and the map allow to
    find which source is pointed by this alias. *)
let select : Syntax.t -> query * Path.t ImportExpression.T.t array =
 fun conf ->
  (* If the filters contains a group expression, we need to transform this into
     a CTE, which have to be evaluated before the main query. That’s why we are
     evaluating the filters right now.*)
  let filters = eval_filters conf in
  let b = Buffer.create 256 in
  let parameters = Queue.create () in

  Option.iter
    (fun (_, (cte : Chunk.t)) ->
      Buffer.add_buffer b cte.b;
      Queue.add_seq parameters (Queue.to_seq cte.parameters))
    filters.cte;

  (* For each column in the configuration file, add the corresponding element
     in the query.

     The Sqlite driver return the elements in an array, we create  an array to
     in order to manage the elements together.
  *)
  let headers = Array.make (List.length conf.columns) (Obj.magic None) in

  let columns = List.to_seq conf.columns |> Seq.mapi (fun i c -> (i, c)) in
  let formatter = Format.formatter_of_buffer b in
  let () =
    Format.fprintf formatter "SELECT %a"
      (Format.pp_print_seq
         ~pp_sep:(fun f () -> Format.fprintf f ",\n")
         (fun formatter (i, column) ->
           Array.set headers i column;
           let p =
             Q.query_of_expression Q.BindParam formatter (show_path ~conf)
               column
           in
           Queue.transfer p parameters;
           Format.fprintf formatter " AS result_%d" i))
      columns
  in
  Format.pp_print_flush formatter ();

  let () = create_from_chunck conf (Chunk.create' b parameters) in

  (* If the query has a CTE, link it as well. We use an INNER JOIN here because
     we want to be sure to get all the rows fetched by the CTE
  *)
  let () =
    match filters.cte with
    | None -> ()
    | Some (name, _) ->
        Buffer.add_string b "\nINNER JOIN '";
        Buffer.add_string b name;
        Buffer.add_string b "' ON ";
        Buffer.add_string b name;
        Buffer.add_string b ".id = ";
        Buffer.add_string b conf.source.name;
        Buffer.add_string b ".id"
  in

  Buffer.add_buffer b filters.content;
  Queue.add_seq parameters filters.parameters;

  let formatter = Format.formatter_of_buffer b in
  (match conf.Syntax.uniq with
  | [] -> ()
  | uniq ->
      Format.fprintf formatter "\nGROUP BY %a"
        (Format.pp_print_list
           ~pp_sep:(fun f () -> Format.fprintf f ", ")
           (fun formatter column ->
             let seq =
               Q.query_of_expression Q.BindParam formatter (show_path ~conf)
                 column
             in
             Queue.transfer seq parameters))
        uniq);
  (match conf.Syntax.sort with
  | [] -> ()
  | sort ->
      Format.fprintf formatter "\nORDER BY %a"
        (Format.pp_print_list
           ~pp_sep:(fun f () -> Format.fprintf f ", ")
           (fun formatter column ->
             let seq =
               Q.query_of_expression Q.BindParam formatter (show_path ~conf)
                 column
             in
             Queue.transfer seq parameters))
        sort);
  Format.pp_print_flush formatter ();

  ({ q = Buffer.contents b; parameters = Queue.to_seq parameters }, headers)

let check_external : Syntax.t -> Syntax.Extern.t -> query =
 fun conf external_ ->
  let internal_chunk =
    let internal_key_buffer = Buffer.create 16 in
    let formatter = Format.formatter_of_buffer internal_key_buffer in
    let internal_key_seq =
      Q.query_of_expression Q.BindParam formatter (show_path ~conf)
        external_.Syntax.Extern.intern_key
    in
    Format.pp_print_flush formatter ();
    Chunk.create' internal_key_buffer (Queue.copy internal_key_seq)
  in

  let external_key_buffer = Buffer.create 16 in
  Buffer.add_string external_key_buffer
    (print_column external_.Syntax.Extern.target
       ("key_" ^ external_.Syntax.Extern.target.name));

  let pointed_tables = pointed_tables conf external_.intern_key in

  let parameters = Queue.create () in
  (* We do a copy before the transfert because the Queue is reused later in the
     query *)
  Queue.transfer (Queue.copy internal_chunk.parameters) parameters;

  (* We have to link all the tables referenced by the external, we cannot let
     any table not linked with the source in the request (this would cause a
     cartesian product request)

     This not the usual way to proceed (we start from the source and link the externals)
   *)
  let rec collect_links :
      Syntax.Extern.t -> Syntax.Extern.t list -> Syntax.Extern.t list =
   fun table init ->
    let res =
      (* Do not add the same external if the value is already present *)
      let init =
        match List.find_opt init ~f:(fun ext -> table == ext) with
        | None -> table :: init
        | Some _ -> init
      in

      Expression.T.fold_values ~init table.Syntax.Extern.intern_key
        ~f:(fun acc expr ->
          match expr.Path.alias with
          | None -> acc
          | Some _ as path -> (
              let table = ImportConf.get_table_for_name conf path in
              (* Look for this table in the externals *)
              let external_opt =
                List.find_opt conf.Syntax.externals ~f:(fun t ->
                    t.Syntax.Extern.target == table)
              in
              match external_opt with
              | None -> acc
              | Some ext -> collect_links ext acc))
    in
    res
  in
  let dependencies = collect_links external_ [] in

  let join_content = Buffer.contents external_key_buffer in
  let request = Chunk.create () in
  Chunk.add_string request "SELECT ";
  let () =
    match pointed_tables with
    | [] ->
        (* Otherwise, just return -1 *)
        Chunk.add_string request "-1"
    | (table, _name) :: _ ->
        (* If we have a single source, extract the row number. *)
        Chunk.add_string request (print_column table "id")
  in
  Chunk.add_string request ", ";
  Chunk.append ~head:request ~tail:(Chunk.copy internal_chunk);
  Chunk.add_string request " FROM\n'";
  Chunk.add_string request (Table.name conf.source);
  Chunk.add_string request "' AS '";
  Chunk.add_string request conf.source.name;
  Chunk.add_string request "'";

  (* Add the externals in the query *)
  List.iter dependencies ~f:(query_of_external ~conf ~join_buffer:request);
  Chunk.add_string request " WHERE ";
  Chunk.add_string request join_content;
  Chunk.add_string request " IS NULL AND ";
  Chunk.append ~head:request ~tail:(Chunk.copy internal_chunk);
  Chunk.add_string request " IS NOT NULL AND ";
  Chunk.append ~head:request ~tail:(Chunk.copy internal_chunk);
  Chunk.add_string request " <> ''";
  let q = Buffer.contents request.b in
  { q; parameters = Queue.to_seq request.parameters }

let build_key_insert : Buffer.t -> Dependency.key -> unit =
 fun buffer { Dependency.expression; _ } ->
  let show_column : Format.formatter -> Path.column -> unit =
   fun formatter column -> Format.fprintf formatter ":col_%d" column
  in

  let formatter = Format.formatter_of_buffer buffer in

  let () =
    prepare_key formatter ~f:(fun formatter ->
        Q.query_of_expression Q.NoParam formatter show_column expression)
  in

  Format.pp_print_flush formatter ()