Allow to use index with or (#1038)

Since #816, mango JSON index on compound fields can be 
selected only if the selector make sure that all the fields 
listed in the index are always present. 

This adds a special case where all clauses of an `$or` can
ensure that a field is present.

For instance, if I had an index:

[A, B]

is_usable would now return true for the selector:

{
  "A": "foo",
  "$or": {
    "B": "bar",
    "B": "baz"
  }
}

but false for:

{
  "A": "foo",
  "$or": {
    "B": "bar",
    "C": "bar"
  }
}

2 files changed, 181 insertions, 7 deletions

diff --git a/src/mango/src/mango_selector.erl b/src/mango/src/mango_selector.erl
index 2a546c9ba..968dc3c74 100644
--- a/src/mango/src/mango_selector.erl
+++ b/src/mango/src/mango_selector.erl
@@ -597,14 +597,31 @@ has_required_fields_int([], Remainder) ->
 has_required_fields_int(Selector, RequiredFields) when not is_list(Selector) ->
     has_required_fields_int([Selector], RequiredFields);
 
-% We can "see" through $and operator. We ignore other
-% combination operators because they can't be used to restrict
-% an index.
+% We can "see" through $and operator. Iterate
+% through the list of child operators.
 has_required_fields_int([{[{<<"$and">>, Args}]}], RequiredFields) 
         when is_list(Args) ->
     has_required_fields_int(Args, RequiredFields);
 
-% Handle $and operator where it has peers
+% We can "see" through $or operator. Required fields
+% must be covered by all children.
+has_required_fields_int([{[{<<"$or">>, Args}]} | Rest], RequiredFields) 
+        when is_list(Args) ->
+    Remainder0 = lists:foldl(fun(Arg, Acc) ->
+        % for each child test coverage against the full
+        % set of required fields
+        Remainder = has_required_fields_int(Arg, RequiredFields),
+
+        % collect the remaining fields across all children
+        Acc ++ Remainder
+    end, [], Args),
+
+    % remove duplicate fields
+    Remainder1 = lists:usort(Remainder0),
+    has_required_fields_int(Rest, Remainder1);
+
+% Handle $and operator where it has peers. Required fields
+% can be covered by any child.
 has_required_fields_int([{[{<<"$and">>, Args}]} | Rest], RequiredFields) 
         when is_list(Args) ->
     Remainder = has_required_fields_int(Args, RequiredFields),
@@ -680,7 +697,8 @@ has_required_fields_nested_and_true_test() ->
           ]
     }]},
 
-    ?assertEqual(true, has_required_fields(Selector, RequiredFields)).
+    Normalized = normalize(Selector),
+    ?assertEqual(true, has_required_fields(Normalized, RequiredFields)).
 
 has_required_fields_and_false_test() ->
     RequiredFields = [<<"A">>, <<"C">>],
@@ -693,7 +711,7 @@ has_required_fields_and_false_test() ->
     Normalized = normalize(Selector),
     ?assertEqual(false, has_required_fields(Normalized, RequiredFields)).
 
-has_required_fields_or_test() ->
+has_required_fields_or_false_test() ->
     RequiredFields = [<<"A">>],
     Selector = {[{<<"$or">>,
           [
@@ -704,4 +722,144 @@ has_required_fields_or_test() ->
     Normalized = normalize(Selector),
     ?assertEqual(false, has_required_fields(Normalized, RequiredFields)).
 
--endif.
\ No newline at end of file
+has_required_fields_or_true_test() ->
+    RequiredFields = [<<"A">>, <<"B">>, <<"C">>],
+    Selector = {[{<<"A">>, "foo"},
+          {<<"$or">>,
+              [
+                  {[{<<"B">>, <<"bar">>}]},
+                  {[{<<"B">>, <<"baz">>}]}
+              ]
+          },
+		  {<<"C">>, "qux"}
+	]},
+    Normalized = normalize(Selector),
+    ?assertEqual(true, has_required_fields(Normalized, RequiredFields)).
+
+has_required_fields_and_nested_or_true_test() ->
+    RequiredFields = [<<"A">>, <<"B">>],
+    Selector1 = {[{<<"$and">>,
+          [
+              {[{<<"A">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector2 = {[{<<"$or">>,
+          [
+              {[{<<"B">>, <<"foo">>}]},
+              {[{<<"B">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector = {[{<<"$and">>,
+          [
+              Selector1,
+              Selector2
+          ]
+    }]},
+    Normalized = normalize(Selector),
+    ?assertEqual(true, has_required_fields(Normalized, RequiredFields)),
+
+    SelectorReverse = {[{<<"$and">>,
+          [
+              Selector2,
+              Selector1
+          ]
+    }]},
+    NormalizedReverse = normalize(SelectorReverse),
+    ?assertEqual(true, has_required_fields(NormalizedReverse, RequiredFields)).
+
+has_required_fields_and_nested_or_false_test() ->
+    RequiredFields = [<<"A">>, <<"B">>],
+    Selector1 = {[{<<"$and">>,
+          [
+              {[{<<"A">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector2 = {[{<<"$or">>,
+          [
+              {[{<<"A">>, <<"foo">>}]},
+              {[{<<"B">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector = {[{<<"$and">>,
+          [
+              Selector1,
+              Selector2
+          ]
+    }]},
+    Normalized = normalize(Selector),
+    ?assertEqual(false, has_required_fields(Normalized, RequiredFields)),
+
+    SelectorReverse = {[{<<"$and">>,
+          [
+              Selector2,
+              Selector1
+          ]
+    }]},
+
+    NormalizedReverse = normalize(SelectorReverse),
+    ?assertEqual(false, has_required_fields(NormalizedReverse, RequiredFields)).
+
+has_required_fields_or_nested_and_true_test() ->
+    RequiredFields = [<<"A">>],
+    Selector1 = {[{<<"$and">>,
+          [
+              {[{<<"A">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector2 = {[{<<"$and">>,
+          [
+              {[{<<"A">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector = {[{<<"$or">>,
+          [
+              Selector1,
+              Selector2
+          ]
+    }]},
+    Normalized = normalize(Selector),
+    ?assertEqual(true, has_required_fields(Normalized, RequiredFields)).
+
+has_required_fields_or_nested_or_true_test() ->
+    RequiredFields = [<<"A">>],
+    Selector1 = {[{<<"$or">>,
+          [
+              {[{<<"A">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector2 = {[{<<"$or">>,
+          [
+              {[{<<"A">>, <<"bar">>}]}
+          ]
+    }]},
+    Selector = {[{<<"$or">>,
+          [
+              Selector1,
+              Selector2
+          ]
+    }]},
+    Normalized = normalize(Selector),
+    ?assertEqual(true, has_required_fields(Normalized, RequiredFields)).
+
+has_required_fields_or_nested_or_false_test() ->
+    RequiredFields = [<<"A">>],
+    Selector1 = {[{<<"$or">>,
+          [
+              {[{<<"A">>, <<"foo">>}]}
+          ]
+    }]},
+    Selector2 = {[{<<"$or">>,
+          [
+              {[{<<"B">>, <<"bar">>}]}
+          ]
+    }]},
+    Selector = {[{<<"$or">>,
+          [
+              Selector1,
+              Selector2
+          ]
+    }]},
+    Normalized = normalize(Selector),
+    ?assertEqual(false, has_required_fields(Normalized, RequiredFields)).
+
+-endif.
diff --git a/src/mango/test/05-index-selection-test.py b/src/mango/test/05-index-selection-test.py
index eec0bd9a6..2a40fda38 100644
--- a/src/mango/test/05-index-selection-test.py
+++ b/src/mango/test/05-index-selection-test.py
@@ -38,6 +38,22 @@ class IndexSelectionTests:
             }, explain=True)
         self.assertEqual(resp["index"]["type"], "json")
 
+    def test_with_or(self):
+        # index on ["company","manager"]
+        ddocid = "_design/a0c425a60cf3c3c09e3c537c9ef20059dcef9198"
+
+        resp = self.db.find({
+                "company": {
+                    "$gt": "a",
+                    "$lt": "z"
+                },
+                "$or": [
+                    {"manager": "Foo"},
+                    {"manager": "Bar"}
+                ]
+            }, explain=True)
+        self.assertEqual(resp["index"]["ddoc"], ddocid)
+
     def test_use_most_columns(self):
         # ddoc id for the age index
         ddocid = "_design/ad3d537c03cd7c6a43cf8dff66ef70ea54c2b40f"