$in queries

3 files changed, 119 insertions, 79 deletions

diff --git a/db/jsobj.cpp b/db/jsobj.cpp
index 50c7e2ae720..8f877c61a38 100644
--- a/db/jsobj.cpp
+++ b/db/jsobj.cpp
@@ -95,6 +95,7 @@ public:
 JSMatcher::~JSMatcher() { 
 	for( int i = 0; i < nBuilders; i++ )
 		delete builders[i];
+	delete in;
 	delete where;
 }
 
@@ -147,7 +148,7 @@ string Element::toString() {
 	return s.str();
 }
 
-int Element::size() {
+int Element::size() const {
 	if( totalSize >= 0 )
 		return totalSize;
 
@@ -194,7 +195,7 @@ int Element::size() {
 			cout << "Element: bad type " << (int) type() << endl;
 			assert(false);
 	}
-	totalSize =  x + fieldNameSize;
+	((Element *) this)->totalSize =  x + fieldNameSize;
 
 	if( !eoo() ) { 
 		const char *next = data + totalSize;
@@ -214,7 +215,7 @@ int Element::size() {
 }
 
 /* must be same type! */
-inline int compareElementValues(Element& l, Element& r) {
+int compareElementValues(const Element& l, const Element& r) {
 	int f;
 	double x;
 	switch( l.type() ) {
@@ -289,15 +290,19 @@ int getGtLtOp(Element& e) {
 
 	Element fe = e.embeddedObject().firstElement();
 	const char *fn = fe.fieldName();
-	if( fn[0] == '$' && fn[1] && fn[2] == 't' ) { 
-		if( fn[1] == 'g' ) { 
-			if( fn[3] == 0 ) op = JSMatcher::GT;
-			else if( fn[3] == 'e' && fn[4] == 0 ) op = JSMatcher::GTE;
-		}
-		else if( fn[1] == 'l' ) { 
-			if( fn[3] == 0 ) op = JSMatcher::LT;
-			else if( fn[3] == 'e' && fn[4] == 0 ) op = JSMatcher::LTE;
+	if( fn[0] == '$' && fn[1] ) { 
+		if( fn[2] == 't' ) { 
+			if( fn[1] == 'g' ) { 
+				if( fn[3] == 0 ) op = JSMatcher::GT;
+				else if( fn[3] == 'e' && fn[4] == 0 ) op = JSMatcher::GTE;
+			}
+			else if( fn[1] == 'l' ) { 
+				if( fn[3] == 0 ) op = JSMatcher::LT;
+				else if( fn[3] == 'e' && fn[4] == 0 ) op = JSMatcher::LTE;
+			}
 		}
+		else if( fn[1] == 'i' && fn[2] == 'n' && fn[3] == 0 )
+			op = JSMatcher::opIN;
 	}
 	return op;
 }
@@ -305,7 +310,7 @@ int getGtLtOp(Element& e) {
 #include "pdfile.h"
 
 JSMatcher::JSMatcher(JSObj &_jsobj) : 
-   where(0), jsobj(_jsobj), nRegex(0)
+   in(0), where(0), jsobj(_jsobj), nRegex(0)
 {
 	nBuilders = 0;
 
@@ -353,26 +358,43 @@ JSMatcher::JSMatcher(JSObj &_jsobj) :
 		}
 
 		// greater than / less than...
-		// { a : { $gt: 3 } }
+		// e.g., e == { a : { $gt : 3 } }
+		//       or 
+		//            { a : { $in : [1,2,3] } }
 		if( e.type() == Object ) {
+			// e.g., fe == { $gt : 3 }
 			Element fe = e.embeddedObject().firstElement();
 			const char *fn = fe.fieldName();
-			if( fn[0] == '$' && fn[1] && fn[2] == 't' ) { 
-				int op = Equality;
-				if( fn[1] == 'g' ) { 
-					if( fn[3] == 0 ) op = GT;
-					else if( fn[3] == 'e' && fn[4] == 0 ) op = GTE;
-				}
-				else if( fn[1] == 'l' ) { 
-					if( fn[3] == 0 ) op = LT;
-					else if( fn[3] == 'e' && fn[4] == 0 ) op = LTE;
+			if( fn[0] == '$' && fn[1] ) { 
+				if( fn[2] == 't' ) { 
+					int op = Equality;
+					if( fn[1] == 'g' ) { 
+						if( fn[3] == 0 ) op = GT;
+						else if( fn[3] == 'e' && fn[4] == 0 ) op = GTE;
+					}
+					else if( fn[1] == 'l' ) { 
+						if( fn[3] == 0 ) op = LT;
+						else if( fn[3] == 'e' && fn[4] == 0 ) op = LTE;
+					}
+					if( op && nBuilders < 8) { 
+						JSObjBuilder *b = new JSObjBuilder();
+						builders[nBuilders++] = b;
+						b->appendAs(fe, e.fieldName());
+						toMatch.push_back( b->done().firstElement() );
+						compareOp.push_back(op);
+						n++;
+						continue;
+					}
 				}
-				if( op && nBuilders < 8) { 
-					JSObjBuilder *b = new JSObjBuilder();
-					builders[nBuilders++] = b;
-					b->appendAs(fe, e.fieldName());
-					toMatch.push_back( b->done().firstElement() );
-					compareOp.push_back(op);
+				else if( fn[1] == 'i' && fn[2] == 'n' && fn[3] == 0 && fe.type() == Array ) {
+					// $in
+					assert( in == 0 ); // only one per query supported so far.  finish...
+					in = new set<Element,element_lt>();
+					JSElemIter i(fe.embeddedObject());
+					while( i.more() )
+						in->insert(i.next());
+					toMatch.push_back(e); // not actually used at the moment
+					compareOp.push_back(opIN);
 					n++;
 					continue;
 				}
@@ -391,15 +413,35 @@ inline int JSMatcher::valuesMatch(Element& l, Element& r, int op) {
 	if( op == 0 ) 
 		return l.valuesEqual(r);
 
+	if( op == opIN ) {
+		// { $in : [1,2,3] }
+		return in->count(l);
+	}
+
+	/* check LT, GTE, ... */
 	if( l.type() != r.type() )
 		return false;
-
 	int c = compareElementValues(l, r);
 	int z = 1 << (c+1); 
 	return (op & z);
 }
 
-/* return value
+/* Check if a particular field matches.
+
+   fieldName - field to match "a.b" if we are reaching into an embedded object.
+   toMatch   - element we want to match.
+   obj       - database object to check against
+   compareOp - Equality, LT, GT, etc.
+   deep      - out param.  set to true/false if we scanned an array
+   isArr     - 
+
+   Special forms:
+
+     { "a.b" : 3 }             means       obj.a.b == 3
+     { a : { $lt : 3 } }       means       obj.a < 3
+	 { a : { $in : [1,2] } }   means       [1,2].contains(obj.a)
+
+   return value
    -1 mismatch
     0 missing element
     1 match
@@ -503,15 +545,6 @@ bool JSMatcher::matches(JSObj& jsobj, bool *deep) {
 	/* assuming there is usually only one thing to match.  if more this
 	could be slow sometimes. */
 
-	for( int r = 0; r < nRegex; r++ ) { 
-		RegexMatcher& rm = regexs[r];
-		Element e = jsobj.getFieldDotted(rm.fieldName);
-		if( e.eoo() )
-			return false;
-		if( !regexMatches(rm, e, deep) )
-			return false;
-	}
-
 	// check normal non-regex cases:
 	for( int i = 0; i < n; i++ ) {
 		Element& m = toMatch[i]; 
@@ -524,35 +557,14 @@ bool JSMatcher::matches(JSObj& jsobj, bool *deep) {
 			return false;
 	}
 
-/*
-		Element e = jsobj.getFieldDotted(m.fieldName(), arrayElName);
-		if( !e.eoo() ) {
-			if( valuesMatch(e, m, compareOp[i]) ) {
-				goto ok;
-			}
-			else if( e.type() == Array ) {
-				JSElemIter ai(e.embeddedObject());
-				while( ai.more() ) { 
-					Element z = ai.next();
-					if( valuesMatch( z, m, compareOp[i]) ) {
-						if( deep )
-							*deep = true;
-						goto ok;
-					}
-				}
-			}
+	for( int r = 0; r < nRegex; r++ ) { 
+		RegexMatcher& rm = regexs[r];
+		Element e = jsobj.getFieldDotted(rm.fieldName);
+		if( e.eoo() )
 			return false;
-		}
-*/
-
-		/* missing.  that is ok iff we were looking for null */
-//		if( m.type() == jstNULL || m.type() == Undefined )
-//			;
-//////		else
-//			return false;
-//ok:
-//		;
-//	}
+		if( !regexMatches(rm, e, deep) )
+			return false;
+	}
 
 	if( where ) { 
 		if( where->func == 0 )
diff --git a/db/jsobj.h b/db/jsobj.h
index c3dc8b8e960..746d211febd 100644
--- a/db/jsobj.h
+++ b/db/jsobj.h
@@ -88,9 +88,9 @@ class Element {
 	friend class JSObj;
 public:
 	string toString();
-	JSType type() { return (JSType) *data; }
-	bool eoo() { return type() == EOO; }
-	int size();
+	JSType type() const { return (JSType) *data; }
+	bool eoo() const { return type() == EOO; }
+	int size() const;
 
 	// wrap this element up as a singleton object.
 	JSObj wrap();
@@ -102,26 +102,27 @@ public:
 
 	// raw data be careful:
 	const char * value() const { return (data + fieldNameSize + 1); }
-	int valuesize() { return size() - fieldNameSize - 1; } 
+	int valuesize() const { return size() - fieldNameSize - 1; } 
 
 	bool boolean() { return *value() ? true : false; }
 
-	unsigned long long date() { return *((unsigned long long*) value()); }
+	unsigned long long date() const { return *((unsigned long long*) value()); }
 	double& number() { return *((double *) value()); }
+	double number() const { return *((double *) value()); }
 	OID& oid() { return *((OID*) value()); }
 
 	// for strings
-	int valuestrsize() { 
+	int valuestrsize() const { 
 		return *((int *) value());
 	}
 
 	// for objects the size *includes* the size of the size field
-	int objsize() { 
+	int objsize() const { 
 		return *((int *) value());
 	}
 
 	// for strings.  also gives you start of the real data for an embedded object
-	const char * valuestr() { return value() + 4; }
+	const char * valuestr() const { return value() + 4; }
 
 	JSObj embeddedObject();
 
@@ -163,14 +164,24 @@ private:
 	}
 	const char *data;
 	int fieldNameSize;
-	int totalSize;
+	int totalSize; /* caches the computed size */
 };
 
+/* l and r MUST have same type when called: check that first. */
+int compareElementValues(const Element& l, const Element& r);
+
 class JSObj {
 	friend class JSElemIter;
 	class Details {
 	public:
-		~Details() { _objdata = 0; }
+        ~Details() {

+			// note refCount means two different things (thus the assert here)

+            assert(refCount <= 0);

+            if (owned()) {

+                free((void *)_objdata);

+            }

+            _objdata = 0;

+        }
 		const char *_objdata;
 		int _objsize;
 		int refCount; // -1 == don't free
@@ -424,8 +435,9 @@ class Where;
    { a : 3 } is the pattern object.
 
    GT/LT:
-   { a : { $gt
+   { a : { $gt : 3 } } 
 
+   TODO: we should rewrite the matcher to be more an AST style.
 */
 class JSMatcher : boost::noncopyable { 
 	int matchesDotted(
@@ -433,13 +445,24 @@ class JSMatcher : boost::noncopyable {
 		Element& toMatch, JSObj& obj, 
 		int compareOp, bool *deep, bool isArr = false);
 
+	struct element_lt

+	{

+		bool operator()(const Element& l, const Element& r) const

+		{

+			int x = (int) l.type() - (int) r.type();
+			if( x < 0 ) return true;
+			if( x > 0 ) return false;
+			return compareElementValues(l,r) < 0;
+		}

+	};
 public:
 	enum { 
 		Equality = 0,
 		LT = 0x1,
 		LTE = 0x3,
 		GTE = 0x6,
-		GT = 0x4 
+		GT = 0x4, 
+		opIN = 0x8 // { x : { $in : [1,2,3] } }
 	};
 
 	static int opDirection(int op) { 
@@ -458,6 +481,7 @@ public:
 private:
 	int valuesMatch(Element& l, Element& r, int op);
 
+	set<Element,element_lt> *in;
 	Where *where;
 	JSObj& jsobj;
 	vector<Element> toMatch;
diff --git a/db/query.cpp b/db/query.cpp
index 1d0221f9df6..0e3ce5cbeb8 100644
--- a/db/query.cpp
+++ b/db/query.cpp
@@ -106,6 +106,10 @@ auto_ptr<Cursor> getIndexCursor(const char *ns, JSObj& query, JSObj& order, bool
 							// compound keys with GT/LT not supported yet via index.
 							goto fail;
 						}
+						if( op >= JSMatcher::opIN ) {
+							// $in does not use an index (at least yet, should when # of elems is tiny)
+							goto fail;
+						}
 						int direction = - JSMatcher::opDirection(op);
 						return auto_ptr<Cursor>( new BtreeCursor(
 							d->indexes[i].head,