Extend document select pruner to generate document select expression

with the precondition that documents are not available.  Attempts to
access document id or document fields results in invalid values
that might enable further pruning of the expression. This expression
can then be used as a filter to skip retrieving documents that fail
the document selection based on attribute vector values for the
document.

4 files changed, 108 insertions, 14 deletions

diff --git a/searchcore/src/tests/proton/common/selectpruner_test.cpp b/searchcore/src/tests/proton/common/selectpruner_test.cpp
index f5621f1c050..a4c47b78e9f 100644
--- a/searchcore/src/tests/proton/common/selectpruner_test.cpp
+++ b/searchcore/src/tests/proton/common/selectpruner_test.cpp
@@ -78,7 +78,8 @@ makeDocTypeRepo()
                      addField("aa", DataType::T_INT).
                      addField("aaa", Array(DataType::T_INT)).
                      addField("aaw", Wset(DataType::T_INT)).
-                     addField("ab", DataType::T_INT));
+                     addField("ab", DataType::T_INT).
+                     addField("ae", DataType::T_INT));
     builder.document(doc_type_id + 1, type_name_2,
                      Struct(header_name_2), Struct(body_name_2).
                      addField("ic", DataType::T_STRING).
@@ -131,6 +132,7 @@ public:
     MockAttributeManager _amgr;
     std::unique_ptr<const DocumentTypeRepo> _repoUP;
     bool _hasFields;
+    bool _hasDocuments;
 
     TestFixture();
 
@@ -156,11 +158,13 @@ public:
 TestFixture::TestFixture()
     : _amgr(),
       _repoUP(),
-      _hasFields(true)
+      _hasFields(true),
+      _hasDocuments(true)
 {
     _amgr.addAttribute("aa", AttributeFactory::createAttribute("aa", { BasicType::INT32 }));
     _amgr.addAttribute("aaa", AttributeFactory::createAttribute("aaa", { BasicType::INT32 , CollectionType::ARRAY}));
     _amgr.addAttribute("aaw", AttributeFactory::createAttribute("aaw", { BasicType::INT32 , CollectionType::WSET}));
+    _amgr.addAttribute("ae", AttributeFactory::createAttribute("ae", { BasicType::INT32 }));
     _repoUP = makeDocTypeRepo();
 }
 
@@ -247,7 +251,7 @@ TestFixture::testPrune(const string &selection,
     ASSERT_TRUE(docType != NULL);
     Document::UP emptyDoc(new Document(*docType, docId));
     emptyDoc->setRepo(repo);
-    SelectPruner pruner(docTypeName, &_amgr, *emptyDoc, repo, _hasFields);
+    SelectPruner pruner(docTypeName, &_amgr, *emptyDoc, repo, _hasFields, _hasDocuments);
     pruner.process(*select);
     std::ostringstream pos;
     pruner.getNode()->print(pos, true, "");
@@ -295,7 +299,7 @@ TEST_F("Test that test setup is OK", TestFixture)
     const DocumentTypeRepo &repo = *f._repoUP;
     const DocumentType *docType = repo.getDocumentType("test");
     ASSERT_TRUE(docType);
-    EXPECT_EQUAL(10u, docType->getFieldCount());
+    EXPECT_EQUAL(11u, docType->getFieldCount());
     EXPECT_EQUAL("String", docType->getField("ia").getDataType().getName());
     EXPECT_EQUAL("String", docType->getField("ib").getDataType().getName());
     EXPECT_EQUAL("Int", docType->getField("aa").getDataType().getName());
@@ -732,6 +736,8 @@ TEST_F("Test that complex field refs are handled", TestFixture)
 {
     f.testPrune("test.ia",
                 "test.ia != null");
+    f.testPrune("test.ia != null",
+                "test.ia != null");
     f.testPrune("test.ia == \"hello\"",
                 "test.ia == \"hello\"");
     f.testPrune("test.ia.foo == \"hello\"",
@@ -762,6 +768,57 @@ TEST_F("Test that complex field refs are handled", TestFixture)
                 "invalid");
     f.testPrune("test.aaw{4} == 4",
                 "test.aaw{4} == 4");
+    f.testPrune("id.namespace == \"hello\"",
+                "id.namespace == \"hello\"");
+    f.testPrune("test.aa == 4 and id.namespace == \"hello\"",
+                "test.aa == 4 and id.namespace == \"hello\"");
+    f.testPrune("test.aa == 4 and test.ae == 5 and id.namespace == \"hello\"",
+                "test.aa == 4 and test.ae == 5 and id.namespace == \"hello\"");
+}
+
+TEST_F("Test that field values are invalid when disabling document access", TestFixture)
+{
+    f._hasDocuments = false;
+    f.testPrune("test.ia",
+                "invalid");
+    f.testPrune("test.ia != null",
+                "invalid");
+    f.testPrune("test.ia == \"hello\"",
+                "invalid");
+    f.testPrune("test.ia.foo == \"hello\"",
+                "invalid");
+    f.testPrune("test.ibs.foo == \"hello\"",
+                "invalid");
+    f.testPrune("test.ibs.x == \"hello\"",
+                "invalid");
+    f.testPrune("test.ia[2] == \"hello\"",
+                "invalid");
+    f.testPrune("test.iba[2] == \"hello\"",
+                "invalid");
+    f.testPrune("test.ia{foo} == \"hello\"",
+                "invalid");
+    f.testPrune("test.ibw{foo} == 4",
+                "invalid");
+    f.testPrune("test.ibw{foo} == \"hello\"",
+                "invalid");
+    f.testPrune("test.ibm{foo} == \"hello\"",
+                "invalid");
+    f.testPrune("test.aa == 4",
+                "test.aa == 4");
+    f.testPrune("test.aa[4] == 4",
+                "invalid");
+    f.testPrune("test.aaa[4] == 4",
+                "invalid");
+    f.testPrune("test.aa{4} == 4",
+                "invalid");
+    f.testPrune("test.aaw{4} == 4",
+                "invalid");
+    f.testPrune("id.namespace == \"hello\"",
+                "invalid");
+    f.testPrune("test.aa == 4 and id.namespace == \"hello\"",
+                "test.aa == 4 and invalid");
+    f.testPrune("test.aa == 4 and test.ae == 5 and id.namespace == \"hello\"",
+                "test.aa == 4 and test.ae == 5 and invalid");
 }
 
 
diff --git a/searchcore/src/vespa/searchcore/proton/common/cachedselect.cpp b/searchcore/src/vespa/searchcore/proton/common/cachedselect.cpp
index c7f935e99ac..8a7bc690e40 100644
--- a/searchcore/src/vespa/searchcore/proton/common/cachedselect.cpp
+++ b/searchcore/src/vespa/searchcore/proton/common/cachedselect.cpp
@@ -161,7 +161,8 @@ CachedSelect::set(const vespalib::string &selection,
                         amgr,
                         emptyDoc,
                         repo,
-                        hasFields);
+                        hasFields,
+                        true);
     pruner.process(*parsed);
     _resultSet = pruner.getResultSet();
     _allFalse = pruner.isFalse();
diff --git a/searchcore/src/vespa/searchcore/proton/common/selectpruner.cpp b/searchcore/src/vespa/searchcore/proton/common/selectpruner.cpp
index da45953e7b9..3e9075607f6 100644
--- a/searchcore/src/vespa/searchcore/proton/common/selectpruner.cpp
+++ b/searchcore/src/vespa/searchcore/proton/common/selectpruner.cpp
@@ -10,6 +10,7 @@
 #include <vespa/document/select/doctype.h>
 #include <vespa/document/select/invalidconstant.h>
 #include <vespa/document/select/valuenodes.h>
+#include <vespa/searchlib/attribute/attributevector.h>
 #include <vespa/searchlib/attribute/iattributemanager.h>
 
 using document::select::And;
@@ -42,6 +43,7 @@ using document::FieldPath;
 using document::Field;
 using document::FieldNotFoundException;
 using search::AttributeGuard;
+using search::attribute::CollectionType;
 
 namespace proton {
 
@@ -49,12 +51,14 @@ SelectPrunerBase::SelectPrunerBase(const vespalib::string &docType,
                                    const search::IAttributeManager *amgr,
                                    const document::Document &emptyDoc,
                                    const document::DocumentTypeRepo &repo,
-                                   bool hasFields)
+                                   bool hasFields,
+                                   bool hasDocuments)
     : _docType(docType),
       _amgr(amgr),
       _emptyDoc(emptyDoc),
       _repo(repo),
-      _hasFields(hasFields)
+      _hasFields(hasFields),
+      _hasDocuments(hasDocuments)
 {
 }
 
@@ -63,7 +67,8 @@ SelectPrunerBase::SelectPrunerBase(const SelectPrunerBase &rhs)
       _amgr(rhs._amgr),
       _emptyDoc(rhs._emptyDoc),
       _repo(rhs._repo),
-      _hasFields(rhs._hasFields)
+      _hasFields(rhs._hasFields),
+      _hasDocuments(rhs._hasDocuments)
 {
 }
 
@@ -71,9 +76,10 @@ SelectPruner::SelectPruner(const vespalib::string &docType,
                            const search::IAttributeManager *amgr,
                            const document::Document &emptyDoc,
                            const document::DocumentTypeRepo &repo,
-                           bool hasFields)
+                           bool hasFields,
+                           bool hasDocuments)
     : CloningVisitor(),
-      SelectPrunerBase(docType, amgr, emptyDoc, repo, hasFields),
+      SelectPrunerBase(docType, amgr, emptyDoc, repo, hasFields, hasDocuments),
       _inverted(false),
       _wantInverted(false),
       _attrFieldNodes(0u)
@@ -371,6 +377,17 @@ SelectPruner::visitFunctionValueNode(const FunctionValueNode &expr)
 
 
 void
+SelectPruner::visitIdValueNode(const IdValueNode &expr)
+{
+    if (!_hasDocuments) {
+        setInvalidVal();
+        return;
+    }
+    CloningVisitor::visitIdValueNode(expr);
+}
+
+
+void
 SelectPruner::visitFieldValueNode(const FieldValueNode &expr)
 {
     if (_docType != expr.getDocType()) {
@@ -378,11 +395,13 @@ SelectPruner::visitFieldValueNode(const FieldValueNode &expr)
         return;
     }
     const document::DocumentType *docType = _repo.getDocumentType(_docType);
+    bool complex = false; // Cannot handle attribute if complex expression
     vespalib::string name(expr.getFieldName());
     for (uint32_t i = 0; i < name.size(); ++i) {
         if (name[i] == '.' || name[i] == '{' || name[i] == '[') {
             // TODO: Check for struct, array, map or weigthed set
             name = expr.getFieldName().substr(0, i);
+            complex = true;
             break;
         }
     }
@@ -417,13 +436,26 @@ SelectPruner::visitFieldValueNode(const FieldValueNode &expr)
     
     _valueNode = expr.clone(); // Replace with different node type for attrs ?
     _valueNode->clearParentheses();
-    ++_fieldNodes;
+    bool svAttr = false;
+    bool attrField = false;
     if (_amgr != nullptr) {
         AttributeGuard::UP ag(_amgr->getAttribute(name));
         if (ag->valid()) {
-            ++_attrFieldNodes;
+            attrField = true;
+            auto av(ag->getSP());
+            if (av->getCollectionType() == CollectionType::SINGLE && !complex) {
+                svAttr = true;
+            }
         }
     }
+    if (!_hasDocuments && !svAttr) {
+        setInvalidVal();
+        return;
+    }
+    ++_fieldNodes;
+    if (attrField) {
+        ++_attrFieldNodes;
+    }
     _priority = FieldValuePriority;
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/common/selectpruner.h b/searchcore/src/vespa/searchcore/proton/common/selectpruner.h
index 8586107398d..4350247d8b3 100644
--- a/searchcore/src/vespa/searchcore/proton/common/selectpruner.h
+++ b/searchcore/src/vespa/searchcore/proton/common/selectpruner.h
@@ -21,13 +21,15 @@ protected:
     const document::Document &_emptyDoc;
     const document::DocumentTypeRepo &_repo;
     bool _hasFields;
+    bool _hasDocuments;
 
 public:
     SelectPrunerBase(const vespalib::string &docType,
                      const search::IAttributeManager *amgr,
                      const document::Document &emptyDoc,
                      const document::DocumentTypeRepo &repo,
-                     bool hasFields);
+                     bool hasFields,
+                     bool hasDocuments);
 
     SelectPrunerBase(const SelectPrunerBase &rhs);
 };
@@ -48,7 +50,8 @@ public:
                  const search::IAttributeManager *amgr,
                  const document::Document &emptyDoc,
                  const document::DocumentTypeRepo &repo,
-                 bool hasFields);
+                 bool hasFields,
+                 bool hasDocuments);
 
     SelectPruner(const SelectPruner *rhs);
     virtual ~SelectPruner();
@@ -70,6 +73,7 @@ private:
     void visitOrBranch(const document::select::Or &expr) override;
     void visitArithmeticValueNode(const document::select::ArithmeticValueNode &expr) override;
     void visitFunctionValueNode(const document::select::FunctionValueNode &expr) override;
+    void visitIdValueNode(const document::select::IdValueNode &expr) override;
     void visitFieldValueNode(const document::select::FieldValueNode &expr) override;
     void invertNode();
     const document::select::Operator &getOperator(const document::select::Operator &op);