Added filters expressions in the externals

3 files changed, 112 insertions, 63 deletions

diff --git a/lib/sql/db.ml b/lib/sql/db.ml
index f2a2653..1d92a2e 100644
--- a/lib/sql/db.ml
+++ b/lib/sql/db.ml
@@ -286,17 +286,6 @@ let create_view : Sqlite3.db -> ImporterSyntax.t -> unit T.result =
 
   Ok ()
 
-(*
-  let query, _ = ImportAnalyser.Query.select output in
-
-  let query =
-    { query with q = Printf.sprintf "CREATE VIEW result AS %s" query.q }
-  in
-  let** statement = execute_query db query in
-
-  Sqlite3.step statement |> T.to_result
-  *)
-
 let check_foreign :
     f:((string * ImportDataTypes.Value.t) array -> unit) ->
     Sqlite3.db ->
@@ -316,66 +305,118 @@ let check_foreign :
   |> T.to_result
 
 let clear_duplicates :
+    is_root:bool ->
     f:((string * ImportDataTypes.Value.t) array -> unit) ->
     'a t ->
     ImportDataTypes.Table.t ->
     ImportAnalyser.Dependency.key list ->
     unit T.result =
- fun ~f db table keys ->
+ fun ~is_root ~f db table keys ->
   let table_name = Table.name table in
 
-  List.fold_left keys ~init:(Ok ())
-    ~f:(fun state { ImportAnalyser.Dependency.name; _ } ->
-      let** _ = state in
-
-      let select_query =
-        String.concat ~sep:""
-          [
-            "SELECT '";
-            table_name;
-            "'.id, '";
-            table_name;
-            "'.'key_";
-            name;
-            "', '";
-            name;
-            "'  FROM '";
-            table_name;
-            "' INNER JOIN (SELECT id, row_number() OVER(PARTITION BY '";
-            table_name;
-            "'.'key_";
-            name;
-            "' ORDER BY (id)) AS row_num from '";
-            table_name;
-            "') other_table WHERE other_table.row_num <> 1 and '";
-            table_name;
-            "'.id = other_table.id";
-          ]
-      in
-      let stmt = Sqlite3.prepare db select_query in
+  let keys_name =
+    List.map keys ~f:(fun ImportAnalyser.Dependency.{ name; filters; _ } ->
+        ( name,
+          filters,
+          String.concat ~sep:"" [ "'"; table_name; "'.'key_"; name; "'" ] ))
+  in
 
-      ignore
-      @@ Sqlite3.iter stmt ~f:(fun data ->
-             let values =
-               Array.mapi data ~f:(fun i value ->
-                   (Sqlite3.column_name stmt i, T.to_datatype value))
-             in
-             f values);
+  let** _ =
+    List.fold_left keys_name ~init:(Ok ())
+      ~f:(fun state (name, filters, key_name) ->
+        let table_external =
+          ImportDataTypes.Table.{ file = ""; tab = 0; name = table_name }
+        in
 
-      let delete_query =
-        Printf.sprintf
-          {|UPDATE '%s'
-SET key_%s = NULL 
+        (* If there are filters to apply in the external, prepare the predicates now *)
+        let join_buffer = ImportAnalyser.Chunk.create () in
+        let () =
+          match filters with
+          | [] -> ()
+          | _ ->
+              ImportAnalyser.Chunk.add_string join_buffer " WHERE ";
+              ImportAnalyser.Chunk.add_expressions ~sep:"\nAND "
+                ~repr:(fun formatter column ->
+                  Format.fprintf formatter "%s"
+                    (Table.print_column table_external
+                       ("col_" ^ string_of_int column)))
+                join_buffer filters
+        in
+        let values =
+          Queue.to_seq join_buffer.parameters
+          |> Seq.map (fun v -> T.of_datatype v)
+          |> List.of_seq
+        in
+
+        let** _ = state in
+
+        let select_query =
+          String.concat ~sep:""
+            [
+              "SELECT '";
+              table_name;
+              "'.id, ";
+              key_name;
+              ", '";
+              name;
+              "'  FROM '";
+              table_name;
+              "' INNER JOIN (SELECT id, row_number() OVER(PARTITION BY ";
+              key_name;
+              " ORDER BY (id)) AS row_num from '";
+              table_name;
+              "'";
+              Buffer.contents join_buffer.b;
+              ") other_table WHERE other_table.row_num <> 1 AND \
+               'other_table'.id = ";
+              Table.print_column table_external "id";
+            ]
+        in
+        let stmt = Sqlite3.prepare db select_query in
+        let* _ = Sqlite3.bind_values stmt values in
+
+        ignore
+        @@ Sqlite3.iter stmt ~f:(fun data ->
+               let values =
+                 Array.mapi data ~f:(fun i value ->
+                     (Sqlite3.column_name stmt i, T.to_datatype value))
+               in
+               f values);
+
+        let clear_query =
+          Printf.sprintf
+            {|UPDATE '%s'
+SET 'key_%s' = NULL 
 FROM 
 (
-	SELECT id, row_number() OVER(PARTITION BY key_%s ORDER BY (id)) AS row_num 
+	SELECT id, row_number() OVER(PARTITION BY %s ORDER BY (id)) AS row_num, * 
 	from '%s'
-) other_table
+%s) other_table
 WHERE other_table.row_num <> 1
 and '%s'.id = other_table.id|}
-          table_name name name table_name table_name
-      in
+            table_name name key_name table_name
+            (Buffer.contents join_buffer.b)
+            table_name
+        in
 
-      Sqlite3.exec db delete_query |> T.to_result)
+        let stmt = Sqlite3.prepare db clear_query in
+        let* _ = Sqlite3.bind_values stmt values in
+        let* _ = Sqlite3.step stmt in
+        Result.ok ())
+  in
+
+  (* Now remove from the database all the line having ALL the keys to null *)
+  match (is_root, keys_name) with
+  | true, _ | _, [] -> Result.ok ()
+  | _ ->
+      let predicates =
+        List.map keys_name ~f:(fun (_, _, key) -> key ^ " IS NULL")
+        |> String.concat ~sep:" AND "
+      in
+      let delete_query =
+        Printf.sprintf {|DELETE FROM '%s' WHERE %s|} table_name predicates
+      in
+      let** _ = Sqlite3.exec db delete_query |> T.to_result in
+      Result.ok ()
 
 type 'a result = ('a, exn) Result.t
diff --git a/lib/sql/db.mli b/lib/sql/db.mli
index 213fb27..b73479b 100644
--- a/lib/sql/db.mli
+++ b/lib/sql/db.mli
@@ -84,12 +84,16 @@ val check_foreign :
   unit result
 
 val clear_duplicates :
+  is_root:bool ->
   f:((string * ImportDataTypes.Value.t) array -> unit) ->
   'a t ->
   ImportDataTypes.Table.t ->
   ImportAnalyser.Dependency.key list ->
   unit result
-(** Remove all duplicated keys in the table by setting them to NULL. *)
+(** Remove all duplicated keys in the table by setting them to NULL.
+
+    This function will check each key referenced used in the table, and will
+    process each key separately.*)
 
 val insert_header :
   'a t ->
diff --git a/lib/sql/hashs.ml b/lib/sql/hashs.ml
index eced2b4..4df8ca1 100644
--- a/lib/sql/hashs.ml
+++ b/lib/sql/hashs.ml
@@ -5,7 +5,6 @@
     before inserting the values. *)
 
 open StdLabels
-module Table = ImportDataTypes.Table
 
 let ( let* ) = Result.bind
 
@@ -21,16 +20,21 @@ let evaluate : ImportAnalyser.Dependency.t -> int =
   (* Extract all the references to this table *)
   let keys =
     List.map (ImportAnalyser.Dependency.keys table)
-      ~f:(fun ImportAnalyser.Dependency.{ name; columns; expression } ->
+      ~f:(fun
+          ImportAnalyser.Dependency.{ name; columns; expression; filters } ->
+        (* It’s better to explicitly ignore the fields we want to exclude than
+           using the pattern _ in the fuction definition. 
+
+           This way, adding a new field in the type will raise a compilation error. *)
         ignore columns;
-        (name, expression))
+        (name, expression, filters))
   in
   Hashtbl.hash keys
 
 let insert : 'a T.t -> ImportAnalyser.Dependency.t -> unit T.result =
  fun db table ->
   let source = ImportAnalyser.Dependency.table table in
-  let table_name = Table.name source in
+  let table_name = ImportDataTypes.Table.name source in
 
   let hash = evaluate table in
 
@@ -56,7 +60,7 @@ let insert : 'a T.t -> ImportAnalyser.Dependency.t -> unit T.result =
 
 let query : 'a T.t -> ImportDataTypes.Table.t -> int option T.result =
  fun db table ->
-  let table_name = Table.name table in
+  let table_name = ImportDataTypes.Table.name table in
   let query =
     String.concat ~sep:""
       [ "SELECT hash FROM 'hashes' WHERE hashes.'table' = '"; table_name; "'" ]