Added comparator chain so that we can merge correctly

stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/QueryProcessor.java

@@ -267,11 +267,12 @@ public Results getResults( final SearchVisitorFactory searchVisitorFactory ) thr
             return null;
         }
 
-        //use the gather iterator to collect all the
+        //use the gather iterator to collect all the          '
+        final int resultSetSize = Math.min( size, Query.MAX_LIMIT );
 
-        ResultIterator itr = new GatherIterator(rootNode, searchVisitorFactory.createVisitors()  );
+        ResultIterator itr = new GatherIterator(resultSetSize, rootNode, searchVisitorFactory.createVisitors()  );
 
-        List<ScanColumn> entityIds = new ArrayList<ScanColumn>( Math.min( size, Query.MAX_LIMIT ) );
+        List<ScanColumn> entityIds = new ArrayList<ScanColumn>( );
 
         CursorCache resultsCursor = new CursorCache();
 

stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/RelationManagerImpl.java

@@ -58,7 +58,9 @@
 import org.apache.usergrid.persistence.hector.CountingMutator;
 import org.apache.usergrid.persistence.query.ir.QuerySlice;
 import org.apache.usergrid.persistence.query.ir.result.CollectionResultsLoaderFactory;
+import org.apache.usergrid.persistence.query.ir.result.CollectionSearchVisitorFactory;
 import org.apache.usergrid.persistence.query.ir.result.ConnectionResultsLoaderFactory;
+import org.apache.usergrid.persistence.query.ir.result.ConnectionSearchVisitorFactory;
 import org.apache.usergrid.persistence.query.ir.result.ConnectionTypesIterator;
 import org.apache.usergrid.persistence.query.ir.result.SearchCollectionVisitor;
 import org.apache.usergrid.persistence.query.ir.result.SearchConnectionVisitor;
@@ -1720,9 +1722,11 @@ public Results searchCollection( String collectionName, Query query ) throws Exc
         // we have something to search with, visit our tree and evaluate the
         // results
         QueryProcessor qp = new QueryProcessor( query, collection, em, factory );
-        SearchCollectionVisitor visitor = new SearchCollectionVisitor( this, qp );
 
-        return qp.getResults( visitor );
+        CollectionSearchVisitorFactory collectionSearchVisitorFactory = new CollectionSearchVisitorFactory( cass, indexBucketLocator, qp, applicationId, headEntity, collectionName );
+//        SearchCollectionVisitor visitor = new SearchCollectionVisitor( this, qp );
+
+        return qp.getResults( collectionSearchVisitorFactory );
     }
 
 
@@ -1914,9 +1918,12 @@ private Results getConnectedEntities( EntityRef sourceEntity, String connectionT
         final ConnectionResultsLoaderFactory factory = new ConnectionResultsLoaderFactory( connectionRef );
 
         QueryProcessor qp = new QueryProcessor( query, null, em, factory );
-        SearchConnectionVisitor visitor = new SearchConnectionVisitor( this, qp, connectionRef, true );
 
-        return qp.getResults( visitor );
+        ConnectionSearchVisitorFactory collectionSearchVisitorFactory = new ConnectionSearchVisitorFactory( cass, indexBucketLocator, qp, applicationId, headEntity, connectionRef, true, "" );
+
+//        SearchConnectionVisitor visitor = new SearchConnectionVisitor( this, qp, connectionRef, true );
+
+        return qp.getResults( collectionSearchVisitorFactory );
     }
 
 
@@ -1956,9 +1963,13 @@ private Results getConnectingEntities(EntityRef targetEntity,
         final ConnectionResultsLoaderFactory factory = new ConnectionResultsLoaderFactory( connectionRef );
 
         QueryProcessor qp = new QueryProcessor( query, null, em, factory );
-        SearchConnectionVisitor visitor = new SearchConnectionVisitor( this, qp, connectionRef, false );
 
-        return qp.getResults( visitor );
+
+        ConnectionSearchVisitorFactory collectionSearchVisitorFactory = new ConnectionSearchVisitorFactory( cass, indexBucketLocator, qp, applicationId, headEntity, connectionRef, false, "" );
+
+//        SearchConnectionVisitor visitor = new SearchConnectionVisitor( this, qp, connectionRef, false );
+
+        return qp.getResults( collectionSearchVisitorFactory );
 	}
 
 
@@ -1995,9 +2006,12 @@ public Results searchConnectedEntities( Query query ) throws Exception {
         final ConnectionResultsLoaderFactory factory = new ConnectionResultsLoaderFactory( connectionRef );
 
         QueryProcessor qp = new QueryProcessor( query, null, em, factory );
-        SearchConnectionVisitor visitor = new SearchConnectionVisitor( this, qp, connectionRef, true );
 
-        return qp.getResults( visitor );
+        ConnectionSearchVisitorFactory collectionSearchVisitorFactory = new ConnectionSearchVisitorFactory( cass, indexBucketLocator, qp, applicationId, headEntity, connectionRef, false, "" );
+
+//        SearchConnectionVisitor visitor = new SearchConnectionVisitor( this, qp, connectionRef, true );
+
+        return qp.getResults( collectionSearchVisitorFactory );
     }
 
 

stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/index/DynamicCompositeComparator.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence.cassandra.index;
+
+
+import java.nio.ByteBuffer;
+import java.util.Comparator;
+
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.TypeParser;
+
+import org.apache.usergrid.persistence.cassandra.ApplicationCF;
+
+
+public abstract class DynamicCompositeComparator implements Comparator<ByteBuffer> {
+    @SuppressWarnings("rawtypes")
+    protected final AbstractType dynamicComposite;
+
+
+    protected DynamicCompositeComparator( ApplicationCF cf ) {
+        // should never happen, this will blow up during development if this fails
+        try {
+            dynamicComposite = TypeParser.parse( cf.getComparator() );
+        }
+        catch ( Exception e ) {
+            throw new RuntimeException( e );
+        }
+    }
+}

stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/index/DynamicCompositeForwardComparator.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence.cassandra.index;
+
+
+import java.nio.ByteBuffer;
+
+import org.apache.usergrid.persistence.cassandra.ApplicationCF;
+
+
+class DynamicCompositeForwardComparator extends DynamicCompositeComparator {
+
+    /**
+     * @param cf
+     */
+    protected DynamicCompositeForwardComparator( ApplicationCF cf ) {
+        super( cf );
+    }
+
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public int compare( ByteBuffer o1, ByteBuffer o2 ) {
+        return dynamicComposite.compare( o1, o2 );
+    }
+}

stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/index/DynamicCompositeReverseComparator.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence.cassandra.index;
+
+
+import java.nio.ByteBuffer;
+
+import org.apache.usergrid.persistence.cassandra.ApplicationCF;
+
+
+class DynamicCompositeReverseComparator extends DynamicCompositeComparator {
+    /**
+     * @param cf
+     */
+    protected DynamicCompositeReverseComparator( ApplicationCF cf ) {
+        super( cf );
+    }
+
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public int compare( ByteBuffer o1, ByteBuffer o2 ) {
+        return dynamicComposite.compare( o2, o1 );
+    }
+}

stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/index/IndexMultiBucketSetLoader.java

@@ -27,9 +27,6 @@
 import org.apache.usergrid.persistence.cassandra.ApplicationCF;
 import org.apache.usergrid.persistence.cassandra.CassandraService;
 
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.TypeParser;
-
 import me.prettyprint.hector.api.beans.HColumn;
 
 
@@ -84,56 +81,4 @@ public int compare( HColumn<ByteBuffer, ByteBuffer> first,
 
         return resultsTree;
     }
-
-
-    private static abstract class DynamicCompositeComparator implements Comparator<ByteBuffer> {
-        @SuppressWarnings("rawtypes")
-        protected final AbstractType dynamicComposite;
-
-
-        protected DynamicCompositeComparator( ApplicationCF cf ) {
-            // should never happen, this will blow up during development if this fails
-            try {
-                dynamicComposite = TypeParser.parse( cf.getComparator() );
-            }
-            catch ( Exception e ) {
-                throw new RuntimeException( e );
-            }
-        }
-    }
-
-
-    private static class DynamicCompositeForwardComparator extends DynamicCompositeComparator {
-
-        /**
-         * @param cf
-         */
-        protected DynamicCompositeForwardComparator( ApplicationCF cf ) {
-            super( cf );
-        }
-
-
-        @SuppressWarnings("unchecked")
-        @Override
-        public int compare( ByteBuffer o1, ByteBuffer o2 ) {
-            return dynamicComposite.compare( o1, o2 );
-        }
-    }
-
-
-    private static class DynamicCompositeReverseComparator extends DynamicCompositeComparator {
-        /**
-         * @param cf
-         */
-        protected DynamicCompositeReverseComparator( ApplicationCF cf ) {
-            super( cf );
-        }
-
-
-        @SuppressWarnings("unchecked")
-        @Override
-        public int compare( ByteBuffer o1, ByteBuffer o2 ) {
-            return dynamicComposite.compare( o2, o1 );
-        }
-    }
 }

stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/index/NoOpIndexScanner.java

@@ -19,8 +19,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Iterator;
-import java.util.NavigableSet;
-import java.util.Set;
+import java.util.List;
 
 import me.prettyprint.hector.api.beans.HColumn;
 
@@ -44,7 +43,7 @@ public NoOpIndexScanner() {
      * @see java.lang.Iterable#iterator()
      */
     @Override
-    public Iterator<Set<HColumn<ByteBuffer, ByteBuffer>>> iterator() {
+    public Iterator<List<HColumn<ByteBuffer, ByteBuffer>>> iterator() {
         return this;
     }
 
@@ -71,7 +70,7 @@ public void reset() {
      * @see java.util.Iterator#next()
      */
     @Override
-    public NavigableSet<HColumn<ByteBuffer, ByteBuffer>> next() {
+    public List<HColumn<ByteBuffer, ByteBuffer>> next() {
         return null;
     }
 

stack/core/src/main/java/org/apache/usergrid/persistence/geo/EntityLocationRef.java

@@ -30,7 +30,7 @@
 import static org.apache.usergrid.utils.StringUtils.stringOrSubstringBeforeFirst;
 
 
-public class EntityLocationRef implements EntityRef {
+public class EntityLocationRef implements EntityRef, Comparable<EntityLocationRef> {
 
     private UUID uuid;
 
@@ -45,10 +45,6 @@ public class EntityLocationRef implements EntityRef {
     private double distance;
 
 
-    public EntityLocationRef() {
-    }
-
-
     public EntityLocationRef( EntityRef entity, double latitude, double longitude ) {
         this( entity.getType(), entity.getUuid(), latitude, longitude );
     }
@@ -224,4 +220,19 @@ else if ( !uuid.equals( other.uuid ) ) {
         }
         return true;
     }
+
+
+    /**
+     * Compares 2 locations by comparing their distance
+     * @param other
+     * @return
+     */
+    @Override
+    public int compareTo( final EntityLocationRef other ) {
+        if(other == null){
+            return 1;
+        }
+
+        return Double.compare( distance, other.distance );
+    }
 }

stack/core/src/main/java/org/apache/usergrid/persistence/query/ir/SearchVisitor.java

@@ -162,7 +162,7 @@ public void visit( OrNode node ) throws Exception {
 
         final int nodeId = node.getId();
 
-        UnionIterator union = new UnionIterator( queryProcessor.getPageSizeHint( node ), nodeId, queryProcessor.getCursorCache(nodeId  ) );
+        UnionIterator union = new UnionIterator( queryProcessor.getPageSizeHint( node ), nodeId, queryProcessor.getCursorCache( nodeId ) );
 
         if ( left != null ) {
             union.addIterator( left );