CDRIVER-3063 add support for pipelines in the update command

Author: isabelatkinson

src/libmongoc/doc/mongoc_collection_update.rst

@@ -22,7 +22,7 @@ Parameters
 * ``collection``: A :symbol:`mongoc_collection_t`.
 * ``flags``: A bitwise or of :symbol:`mongoc_update_flags_t`.
 * ``selector``: A :symbol:`bson:bson_t` containing the query to match documents for updating.
-* ``update``: A :symbol:`bson:bson_t` containing the update to perform.
+* ``update``: A :symbol:`bson:bson_t` containing the update to perform. If updating with a pipeline, a :symbol:`bson:bson_t` array.
 * ``write_concern``: A :symbol:`mongoc_write_concern_t`.
 * ``error``: An optional location for a :symbol:`bson_error_t <errors>` or ``NULL``.
 

src/libmongoc/doc/mongoc_collection_update_many.rst

@@ -21,7 +21,7 @@ Parameters
 
 * ``collection``: A :symbol:`mongoc_collection_t`.
 * ``selector``: A :symbol:`bson:bson_t` containing the query to match documents for updating.
-* ``update``: A :symbol:`bson:bson_t` containing the update to perform.
+* ``update``: A :symbol:`bson:bson_t` containing the update to perform. If updating with a pipeline, a :symbol:`bson:bson_t` array.
 * ``reply``: Optional. An uninitialized :symbol:`bson:bson_t` populated with the update result, or ``NULL``.
 * ``error``: An optional location for a :symbol:`bson_error_t <errors>` or ``NULL``.
 

src/libmongoc/doc/mongoc_collection_update_one.rst

@@ -21,7 +21,7 @@ Parameters
 
 * ``collection``: A :symbol:`mongoc_collection_t`.
 * ``selector``: A :symbol:`bson:bson_t` containing the query to match the document for updating.
-* ``update``: A :symbol:`bson:bson_t` containing the update to perform.
+* ``update``: A :symbol:`bson:bson_t` containing the update to perform. If updating with a pipeline, a :symbol:`bson:bson_t` array.
 * ``reply``: Optional. An uninitialized :symbol:`bson:bson_t` populated with the update result, or ``NULL``.
 * ``error``: An optional location for a :symbol:`bson_error_t <errors>` or ``NULL``.
 

src/libmongoc/src/mongoc/mongoc-collection.c

@@ -1753,7 +1753,6 @@ mongoc_collection_insert_many (mongoc_collection_t *collection,
    RETURN (ret);
 }
 
-
 /*
  *--------------------------------------------------------------------------
  *
@@ -3066,7 +3065,11 @@ mongoc_collection_find_and_modify_with_opts (
    }
 
    if (opts->update) {
-      BSON_APPEND_DOCUMENT (&command, "update", opts->update);
+      if (_mongoc_document_is_pipeline (opts->update)) {
+         BSON_APPEND_ARRAY (&command, "update", opts->update);
+      } else {
+         BSON_APPEND_DOCUMENT (&command, "update", opts->update);
+      }
    }
 
    if (opts->fields) {

src/libmongoc/src/mongoc/mongoc-util-private.h

@@ -131,6 +131,9 @@ void
 _mongoc_bson_init_with_transient_txn_error (const mongoc_client_session_t *cs,
                                             bson_t *reply);
 
+bool
+_mongoc_document_is_pipeline (const bson_t *document);
+
 BSON_END_DECLS
 
 #endif /* MONGOC_UTIL_PRIVATE_H */

src/libmongoc/src/mongoc/mongoc-util.c

@@ -363,6 +363,10 @@ _mongoc_validate_update (const bson_t *update,
       return false;
    }
 
+   if (_mongoc_document_is_pipeline (update)) {
+      return true;
+   }
+
    if (!bson_iter_init (&iter, update)) {
       bson_set_error (error,
                       MONGOC_ERROR_BSON,
@@ -377,7 +381,8 @@ _mongoc_validate_update (const bson_t *update,
          bson_set_error (error,
                          MONGOC_ERROR_COMMAND,
                          MONGOC_ERROR_COMMAND_INVALID_ARG,
-                         "Invalid key '%s': update only works with $ operators",
+                         "Invalid key '%s': update only works with $ operators"
+                         " and pipelines",
                          key);
 
          return false;
@@ -521,3 +526,47 @@ _mongoc_bson_init_with_transient_txn_error (const mongoc_client_session_t *cs,
       bson_append_array_end (reply, &labels);
    }
 }
+
+bool
+_mongoc_document_is_pipeline (const bson_t *document)
+{
+   bson_iter_t iter;
+   bson_iter_t child;
+   const char *key;
+   int i = 0;
+   char *i_str;
+
+   if (!bson_iter_init (&iter, document)) {
+      return false;
+   }
+
+   while (bson_iter_next (&iter)) {
+      key = bson_iter_key (&iter);
+      i_str = bson_strdup_printf ("%d", i++);
+
+      if (strcmp (key, i_str)) {
+         bson_free (i_str);
+         return false;
+      }
+
+      bson_free (i_str);
+
+      if (BSON_ITER_HOLDS_DOCUMENT (&iter)) {
+         if (!bson_iter_recurse (&iter, &child)) {
+            return false;
+         }
+         if (!bson_iter_next (&child)) {
+            return false;
+         }
+         key = bson_iter_key (&child);
+         if (key[0] != '$') {
+            return false;
+         }
+      } else {
+         return false;
+      }
+   }
+
+   /* should return false when the document is empty */
+   return i != 0;
+}

src/libmongoc/src/mongoc/mongoc-write-command.c

@@ -118,7 +118,11 @@ _mongoc_write_command_update_append (mongoc_write_command_t *command,
 
    bson_init (&document);
    BSON_APPEND_DOCUMENT (&document, "q", selector);
-   BSON_APPEND_DOCUMENT (&document, "u", update);
+   if (_mongoc_document_is_pipeline (update)) {
+      BSON_APPEND_ARRAY (&document, "u", update);
+   } else {
+      BSON_APPEND_DOCUMENT (&document, "u", update);
+   }
    if (opts) {
       bson_concat (&document, opts);
    }

src/libmongoc/tests/json/crud/v2/updateWithPipelines.json

@@ -22,7 +22,6 @@
   "tests": [
     {
       "description": "UpdateOne using pipelines",
-      "skipReason": "blocked on CDRIVER-3063",
       "operations": [
         {
           "name": "updateOne",
@@ -101,7 +100,6 @@
     },
     {
       "description": "UpdateMany using pipelines",
-      "skipReason": "blocked on CDRIVER-3063",
       "operations": [
         {
           "name": "updateMany",
@@ -175,7 +173,6 @@
     },
     {
       "description": "FindOneAndUpdate using pipelines",
-      "skipReason": "blocked on CDRIVER-3063",
       "operations": [
         {
           "name": "findOneAndUpdate",

src/libmongoc/tests/test-mongoc-collection.c

@@ -1286,6 +1286,73 @@ test_update (void)
    mongoc_client_destroy (client);
 }
 
+static void
+test_update_pipeline (void *ctx)
+{
+   mongoc_collection_t *collection;
+   mongoc_database_t *database;
+   mongoc_client_t *client;
+   bson_error_t error;
+   bson_t *b;
+   bson_t *pipeline;
+   bson_t *replacement;
+   bool res;
+
+   client = test_framework_client_new ();
+   ASSERT (client);
+
+   database = get_test_database (client);
+   ASSERT (database);
+
+   collection = get_test_collection (client, "test_update_pipeline");
+   ASSERT (collection);
+
+   b = tmp_bson ("{'nums': {'x': 1, 'y': 2}}");
+   res = mongoc_collection_insert_one (collection, b, NULL, NULL, &error);
+   ASSERT_OR_PRINT (res, error);
+
+   /* format: array document with incrementing keys
+      (i.e. {"0": value, "1": value, "2": value}) */
+   pipeline = tmp_bson ("{'0': {'$replaceRoot': {'newRoot': '$nums'}},"
+                        " '1': {'$addFields': {'z': 3}}}");
+   res = mongoc_collection_update_one (
+      collection, b, pipeline, NULL, NULL, &error);
+   ASSERT_OR_PRINT (res, error);
+
+   res = mongoc_collection_insert_one (collection, b, NULL, NULL, &error);
+   ASSERT_OR_PRINT (res, error);
+
+   /* ensure that arrays sent to mongoc_collection_replace_one are not
+      treated as pipelines */
+   replacement = tmp_bson ("{'0': 0, '1': 1}");
+   res = mongoc_collection_replace_one (
+      collection, b, replacement, NULL, NULL, &error);
+   ASSERT_OR_PRINT (res, error);
+
+   /* ensure that pipeline updates sent to mongoc_collection_replace_one
+      receive a client-side error */
+   res = mongoc_collection_replace_one (
+      collection, b, pipeline, NULL, NULL, &error);
+   ASSERT (!res);
+   ASSERT_ERROR_CONTAINS (error,
+                          MONGOC_ERROR_COMMAND,
+                          MONGOC_ERROR_COMMAND_INVALID_ARG,
+                          "invalid argument for replace");
+
+   /* ensure that a pipeline with an empty document is considered invalid */
+   pipeline = tmp_bson ("{ '0': {} }");
+   res = mongoc_collection_update_one (
+      collection, b, pipeline, NULL, NULL, &error);
+   ASSERT (!res);
+   ASSERT_ERROR_CONTAINS (error,
+                          MONGOC_ERROR_COMMAND,
+                          MONGOC_ERROR_COMMAND_INVALID_ARG,
+                          "Invalid key");
+
+   mongoc_collection_destroy (collection);
+   mongoc_database_destroy (database);
+   mongoc_client_destroy (client);
+}
 
 static void
 test_update_oversize (void *ctx)
@@ -6175,6 +6242,12 @@ test_collection_install (TestSuite *suite)
                       NULL,
                       skip_unless_server_has_decimal128);
    TestSuite_AddLive (suite, "/Collection/update", test_update);
+   TestSuite_AddFull (suite,
+                      "/Collection/update_pipeline",
+                      test_update_pipeline,
+                      NULL,
+                      NULL,
+                      test_framework_skip_if_max_wire_version_less_than_8);
    TestSuite_AddLive (suite, "/Collection/update/multi", test_update_multi);
    TestSuite_AddLive (suite, "/Collection/update/upsert", test_update_upsert);
    TestSuite_AddFull (suite,
@@ -6346,4 +6419,4 @@ test_collection_install (TestSuite *suite)
                       NULL,
                       NULL,
                       test_framework_skip_if_not_replset);
-}
+}