initial aggregation framework and bug fix in cursor response

c_wrapper/lua-bson.c

@@ -57,48 +57,16 @@ _convert_to_absolute_stack_index(lua_State *L,
  * @L: A lua_State.
  * @index: An int.
  *
- * Takes a table at indice on stack and iterates through it. If it only has
- * ascending number indices starting at 1, then it is an array. If not, it is
+ * Takes a table at index on stack and iterates through it. If it only has
+ * ascending number indexes starting at 1, then it is an array. If not, it is
  * a table.
  */
 
 bool
 lua_table_is_array(lua_State *L, int index)
 {
-    int num_keys;
     int absolute_stack_index = _convert_to_absolute_stack_index(L, index);
-
-    // Iterate through keys and check if they are all numbers.
-    lua_pushnil(L);
-
-    for (num_keys = 0; lua_next(L, -2) != 0; num_keys++) {
-        if ((lua_type(L, -2)) != LUA_TNUMBER) {
-            lua_pop(L, 2);
-            return false;
-        }
-        lua_pop(L, 1);
-    }
-
-    // Empty table defaults to an empty table instead of empty array.
-    if (num_keys == 0) {
-        return false;
-    }
-
-    // Iterate through like ipairs, and make sure the indices are in ascending
-    // order and there are no gaps.
-    for (int i=1 ; i < num_keys; i++) {
-        lua_rawgeti(L, absolute_stack_index, i);
-
-        // If the index does not exist, it will cause lua_isnil to return nil.
-        if ( lua_isnil(L,-1) ) {
-            lua_pop(L,1);
-            return false;
-        }
-
-        lua_pop(L,1);
-    }
-
-    return true;
+    return lua_array_length(L, absolute_stack_index) > 0;
 }
 
 int
@@ -199,7 +167,7 @@ _append_to_bson_doc (lua_State *L,
             } else {
 
                 bson_t subdocument;
-                if (lua_array_length(L, -1) > 0) {
+                if (lua_table_is_array(L, -1)) {
                     bson_append_array_begin(bson_doc, key, -1, &subdocument);
                     if (!(lua_table_to_bson(L, &subdocument, -1, false, error))) {
                         return false;
@@ -306,7 +274,14 @@ lua_table_to_bson(lua_State *L,
                   bool _id_required,
                   bson_error_t *error)
 {
+    // index: relative or absolute position of table on the stack being
+    // converted
+    // absolute_stack_index: index converted to an absolute stack index
+    // array_index: Potentially altered lua index (since lua indexes start at 1)
+
     int absolute_stack_index = _convert_to_absolute_stack_index(L, index);
+    bool is_array;
+    int array_index;
 
     if(!lua_istable(L, absolute_stack_index)) {
         bson_snprintf(error->message, sizeof(error->message),
@@ -329,6 +304,8 @@ lua_table_to_bson(lua_State *L,
         }
     }
 
+    is_array = lua_table_is_array(L, absolute_stack_index);
+
     lua_pushnil(L);
     while (lua_next(L, absolute_stack_index) != 0) {
         bool number_string_as_index = false;
@@ -350,7 +327,15 @@ lua_table_to_bson(lua_State *L,
                     // Function tostring converts the value on the stack.
                     // Need to make copy, convert it, then pop to allow lua_next
                     // to iterate correctly.
-                    lua_pushvalue(L, -2);
+                    array_index = lua_tonumber(L, -2);
+
+                    // Lua indices start at 1, so decrement by one if it is
+                    // an array.
+                    if (is_array) {
+                        array_index--;
+                    }
+
+                    lua_pushnumber(L, array_index);
                     key = lua_tostring(L, -1);
                     lua_pop(L, 1);
                 }

c_wrapper/lua-bson.h

@@ -21,6 +21,8 @@
 #include "lua-mongoc-wrapper.h"
 #include "lua-object-generators.h"
 
+bool lua_table_is_array(lua_State *L, int index);
+
 bool find_and_set_or_create_id(lua_State *L,
                                int index,
                                bson_t *bson_doc,

c_wrapper/lua-mongo-cursor.c

@@ -52,7 +52,7 @@ lua_mongo_cursor_iterate(lua_State *L)
     cursor = (lua_mongo_cursor_t *)luaL_checkudata(L, 1, "lua_mongo_cursor");
 
     if (mongoc_cursor_next(cursor->c_cursor, &doc)) {
-        if (!(bson_document_or_array_to_table (L, doc, false, &error))) {
+        if (!(bson_document_or_array_to_table (L, doc, true, &error))) {
             throw_error = true;
         }
 

c_wrapper/lua-mongoc-collection.c

@@ -442,6 +442,53 @@ lua_mongo_collection_insert_many (lua_State *L)
     return 1;
 }
 
+
+int lua_mongo_collection_aggregate (lua_State *L)
+{
+    collection_t *collection;
+    bson_t aggregation_pipeline = BSON_INITIALIZER;
+    bson_t inner_aggregation_pipeline = BSON_INITIALIZER;
+    mongoc_cursor_t *cursor = NULL;
+    bson_error_t error;
+    bool throw_error = false;
+
+    collection = (collection_t *)luaL_checkudata(L, 1, "lua_mongoc_collection");
+
+    if (!(lua_istable(L, 2)) || !(lua_table_is_array(L, 2))) {
+        luaL_error(L, "aggregation pipeline must be an array");
+    } else {
+        bson_append_array_begin(&aggregation_pipeline, "pipeline", -1,
+                                &inner_aggregation_pipeline);
+
+        if (!(lua_table_to_bson(L, &inner_aggregation_pipeline, 2, false, &error))) {
+            throw_error = true;
+            goto DONE;
+        }
+
+        bson_append_array_end(&aggregation_pipeline, &inner_aggregation_pipeline);
+    }
+
+    cursor = mongoc_collection_aggregate (collection->c_collection, MONGOC_QUERY_NONE,
+                                          &aggregation_pipeline,
+                                          NULL, NULL);
+
+DONE:
+    bson_destroy(&aggregation_pipeline);
+    bson_destroy(&inner_aggregation_pipeline);
+
+    if (throw_error) {
+        if (cursor) {
+            mongoc_cursor_destroy (cursor);
+        }
+        luaL_error(L, error.message);
+    }
+
+    lua_mongo_cursor_new(L, cursor);
+
+    return 1;
+}
+
+
 int
 lua_mongo_collection_delete_one(lua_State *L)
 {

c_wrapper/lua-mongoc-collection.h

@@ -32,6 +32,7 @@ int lua_mongo_collection_insert_many (lua_State *L);
 int lua_mongo_collection_update (lua_State *L);
 int lua_mongo_collection_delete_one(lua_State *L);
 int lua_mongo_collection_delete_many(lua_State *L);
+int lua_mongo_collection_aggregate (lua_State *L);
 int lua_mongo_collection_destroy (lua_State *L);
 
 static const struct luaL_Reg lua_mongoc_collection_methods[] = {
@@ -45,6 +46,7 @@ static const struct luaL_Reg lua_mongoc_collection_methods[] = {
     { "collection_update_many", lua_mongo_collection_update },
     { "collection_delete_one", lua_mongo_collection_delete_one },
     { "collection_delete_many", lua_mongo_collection_delete_many },
+    { "collection_aggregate", lua_mongo_collection_aggregate },
     { "__gc", lua_mongo_collection_destroy },
     { NULL, NULL },
 };

c_wrapper/lua-object-generators.c

@@ -36,7 +36,7 @@ generate_ObjectID(lua_State *L,
     lua_getglobal(L, "ObjectId");
     if (!lua_istable(L, -1)) {
         strncpy (error->message,
-                 "ObjectId could not be imported",
+                 "ObjectId not a global variable",
                  sizeof (error->message));
         return false;
     }
@@ -159,7 +159,7 @@ generate_InsertOneResult(lua_State *L,
     lua_getglobal(L, "InsertOneResult");
     if (!lua_istable(L, -1)) {
         strncpy (error->message,
-                 "InsertOneResult could not be imported",
+                 "InsertOneResult not a global variable",
                  sizeof (error->message));
         return false;
     }
@@ -219,7 +219,7 @@ generate_InsertManyResult(lua_State *L,
 
     lua_getglobal(L, "InsertManyResult");
     if (!lua_istable(L, -1)) {
-        strncpy (error->message, "InsertManyResult could not be imported",
+        strncpy (error->message, "InsertManyResult not a global variable",
                  sizeof (error->message));
         return false;
     }
@@ -276,7 +276,7 @@ generate_DeleteResult(lua_State *L,
     lua_getglobal(L, "DeleteResult");
     if (!lua_istable(L, -1)) {
         strncpy (error->message,
-                 "DeleteResult could not be imported",
+                 "DeleteResult not a global variable",
                  sizeof (error->message));
         return false;
     }
@@ -309,4 +309,4 @@ generate_DeleteResult(lua_State *L,
     lua_remove(L, -2);
 
     return true;
-}
+}

luaMongo/MongoCollection.lua

@@ -54,7 +54,7 @@ MongoCollection.__index = MongoCollection
 	end
 
 	function MongoCollection:find(query, fields)
-		cursor_t = self.collection_t:collection_find(self, query, fields)
+		local cursor_t = self.collection_t:collection_find(self, query, fields)
 		return createCursorIterator(self, cursor_t)
 	end
 
@@ -91,8 +91,9 @@ MongoCollection.__index = MongoCollection
 		return self.collection_t:collection_delete_many(selector)
 	end
 
-	function MongoCollection:makeRandomObjectId()
-		return self.collection_t:make_random_object_id()
+	function MongoCollection:aggregate(aggregationPipeline)
+		local cursor_t = self.collection_t:collection_aggregate(aggregationPipeline)
+		return createCursorIterator(self, cursor_t)
 	end
-	
+
 return MongoCollection

test/test_collection.lua

@@ -201,7 +201,7 @@ TestClient = {}
 		local client = MongoClient.new("mongodb://user:password@localhost:27017/?authSource=admin")
 		local database = client:getDatabase("foo")
 		local collection = database:getCollection("test")
-
+		
 		collection:drop()
 		collection = database:getCollection("test")
 
@@ -228,51 +228,6 @@ TestClient = {}
 		assert(result.modified_count == 0)
 		assert(ObjectId.isObjectId(result.upserted_id))
 	end
-
-	function TestClient:test_delete_one()
-		local client = MongoClient.new("mongodb://user:password@localhost:27017/?authSource=admin")
-		local database = client:getDatabase("foo")
-		local collection = database:getCollection("test")
-
-		collection:drop()
-		collection = database:getCollection("test")
-
-		collection:insert_one({x = 1})
-		collection:insert_one({y = 1})
-		collection:insert_one({z = 1})
-
-		local result = collection:delete_one({x = 1})
-		assert(DeleteResult.isDeleteResult(result))
-		assert(result.deleted_count == 1)
-		assert(result.acknowledged)
-		assert(collection:count() == 2)
-
-		result = collection:delete_one({y = 1})
-		assert(DeleteResult.isDeleteResult(result))
-		assert(result.deleted_count == 1)
-		assert(result.acknowledged)
-		assert(collection:count() == 1)
-	end
-
-	function TestClient:test_delete_many()
-		local client = MongoClient.new("mongodb://user:password@localhost:27017/?authSource=admin")
-		local database = client:getDatabase("foo")
-		local collection = database:getCollection("test")
-
-		collection:drop()
-		collection = database:getCollection("test")
-
-		collection:insert_one({x = 1})
-		collection:insert_one({x = 1})
-		collection:insert_one({y = 1})
-		collection:insert_one({y = 1})
-
-		local result = collection:delete_many({x = 1})
-		assert(DeleteResult.isDeleteResult(result))
-		assert(result.deleted_count == 2)
-		assert(result.acknowledged)
-		assert(collection:count() == 2)
-	end
 
 	function TestClient:test_count()
 		local client = MongoClient.new("mongodb://user:password@localhost:27017/?authSource=admin")
@@ -294,4 +249,28 @@ TestClient = {}
 		assert(collection:count({foo = {["$regex"] = "ba.*"}}) == 2)
 	end
 
+	function TestClient:test_aggregate()
+		local client = MongoClient.new("mongodb://user:password@localhost:27017/?authSource=admin")
+		local database = client:getDatabase("foo")
+		local collection = database:getCollection("test")
+
+		collection:drop()
+		local inserted_document = { foo = {1, 2} }
+		collection:insert_one(inserted_document)
+		inserted_document["_id"] = nil
+		 local aggregationPipeline = { {["$project"] = {_id = false, foo = true}} }
+
+		local results = collection:aggregate(aggregationPipeline)
+
+		local result_array = {}
+		local index = 1
+		for result in results do
+			result_array[index] = result
+			index = index + 1
+		end
+
+		assert(#result_array == 1)
+		assert(table_eq(inserted_document, result_array[1]))
+	end
+
 LuaUnit:run()