Fixed a bug when a group function did not partition in the same way as the uniq parameter

1 files changed, 71 insertions, 10 deletions

diff --git a/lib/configuration/read_conf.ml b/lib/configuration/read_conf.ml
index 52ccb62..69240c1 100644
--- a/lib/configuration/read_conf.ml
+++ b/lib/configuration/read_conf.ml
@@ -126,6 +126,45 @@ end = struct
   let column = Expression_parser.Incremental.column_expr
 end
 
+(** Ensure the group criteria in window functions match the global group by
+    criteria.
+
+    Traverse the configuration tree until finding a group window. *)
+let matchWindowGroup :
+    eq:('a -> 'a -> bool) ->
+    subset:'a ImportExpression.T.t list ->
+    'a ImportExpression.T.t ->
+    bool =
+ fun ~eq ~subset expression ->
+  let exception Divergent in
+  let open ImportExpression.T in
+  let rec f = function
+    | Empty | Literal _ | Integer _ | Path _ -> ()
+    | Expr e -> f e
+    | Concat pp | Function' (_, pp) | Function (_, pp) | Nvl pp | Join (_, pp)
+      -> List.iter ~f pp
+    | Window (_, pp1, _) ->
+        if List.equal ~eq:(ImportExpression.T.equal eq) subset pp1 then ()
+        else raise_notrace Divergent
+    | BOperator (_, arg1, arg2) ->
+        f arg1;
+        f arg2
+    | GEquality (_, arg1, args) ->
+        f arg1;
+        List.iter ~f args
+  in
+  match subset with
+  | [] ->
+      (* Do not bother traversing the tree if there is no group by, just
+         return Ok *)
+      true
+  | _ -> (
+      try
+        f expression;
+        true
+      with
+      | Divergent -> false)
+
 module Make (S : Decoders.Decode.S) = struct
   let ( let* ) = S.( let* )
   let ( and* ) = S.( and* )
@@ -135,14 +174,24 @@ module Make (S : Decoders.Decode.S) = struct
   class loader =
     object (self)
       method parse_expression : type a.
+          ?groups:a ImportExpression.T.t list ->
+          eq:(a -> a -> bool) ->
           a ExpressionParser.path_builder ->
           S.value ->
           (a ImportExpression.T.t, S.value Decoders.Error.t) result =
-        fun path ->
+        fun ?(groups = []) ~eq path ->
           S.string >>= fun v ->
           match ExpressionParser.of_string path v with
-          | Ok expr -> S.succeed expr
           | Error e -> S.fail_with Decoders.Error.(make e)
+          | Ok expr -> (
+              (* Now check that every window function include at least the uniq list *)
+              let valid_subset = matchWindowGroup ~eq ~subset:groups expr in
+              match valid_subset with
+              | true -> S.succeed expr
+              | false ->
+                  S.fail
+                    "The group function shall match the same arguments as the \
+                     \"uniq\" parameter")
 
       method source =
         let* file = S.field "file" S.string
@@ -152,9 +201,11 @@ module Make (S : Decoders.Decode.S) = struct
 
       method external_ name =
         let* intern_key =
-          S.field "intern_key" (self#parse_expression ExpressionParser.path)
+          S.field "intern_key"
+            (self#parse_expression ~eq:Path.equal ExpressionParser.path)
         and* extern_key =
-          S.field "extern_key" (self#parse_expression ExpressionParser.column)
+          S.field "extern_key"
+            (self#parse_expression ~eq:Int.equal ExpressionParser.column)
         and* file = S.field "file" S.string
         and* tab = S.field_opt_or ~default:1 "tab" S.int
         and* allow_missing =
@@ -172,18 +223,28 @@ module Make (S : Decoders.Decode.S) = struct
             }
 
       method sheet =
+        (* Check the uniq property first, beecause the group functions need
+             to include the same expression (at least) *)
+        let* uniq =
+          S.field_opt_or ~default:[] "uniq"
+          @@ S.list (self#parse_expression ~eq:Path.equal ExpressionParser.path)
+        in
+
         let* columns =
           S.field "columns"
-          @@ S.list (self#parse_expression ExpressionParser.path)
+          @@ S.list
+               (self#parse_expression ~eq:Path.equal ~groups:uniq
+                  ExpressionParser.path)
         and* filters =
           S.field_opt_or ~default:[] "filters"
-          @@ S.list (self#parse_expression ExpressionParser.path)
+          @@ S.list
+               (self#parse_expression ~eq:Path.equal ~groups:uniq
+                  ExpressionParser.path)
         and* sort =
           S.field_opt_or ~default:[] "sort"
-          @@ S.list (self#parse_expression ExpressionParser.path)
-        and* uniq =
-          S.field_opt_or ~default:[] "uniq"
-          @@ S.list (self#parse_expression ExpressionParser.path)
+          @@ S.list
+               (self#parse_expression ~eq:Path.equal ~groups:uniq
+                  ExpressionParser.path)
         in
         S.succeed @@ fun version source externals locale ->
         Syntax.