CDRIVER-4277: Pre-point-in-time support (#1016)

- Support NULLable strings in generate-opts:
- If an option is given a "None" default value, it's C value will be NULL.
- Updates the `fullDocument` as well as add the `fullDocumentBeforeChange` option.
- Forward the `fullDocumentBeforeChange` option through changestream
options. 
- Enable the pre_and_post_images tests.

Author: vector-of-bool

build/generate-opts.py

@@ -13,7 +13,6 @@
 # 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.
-
 """IDL for functions that take flexible options as a bson_t.
 
 Defines the options accepted by functions that receive a const bson_t *opts,
@@ -25,6 +24,7 @@
 Written for Python 2.6+, requires Jinja 2 for templating.
 """
 
+# yapf: disable
 from collections import OrderedDict
 from os.path import basename, dirname, join as joinpath, normpath
 import re
@@ -292,9 +292,25 @@ def __init__(self, items, **defaults):
         ('startAfter', {'type': 'document', 'help': 'A ``Document`` representing the logical starting point of the change stream. Unlike ``resumeAfter``, this can resume notifications after an "invalidate" event. The result of :symbol:`mongoc_change_stream_get_resume_token()` or the ``_id`` field  of any change received from a change stream can be used here.  This option is mutually exclusive with ``resumeAfter`` and ``startAtOperationTime``.'}),
         ('startAtOperationTime', {'type': 'timestamp', 'help': 'A ``Timestamp``. The change stream only provides changes that occurred at or after the specified timestamp. Any command run against the server will return an operation time that can be used here. This option is mutually exclusive with ``resumeAfter`` and ``startAfter``.'}),
         ('maxAwaitTimeMS', {'type': 'int64_t', 'convert': '_mongoc_convert_int64_positive', 'help': 'An ``int64`` representing the maximum amount of time a call to :symbol:`mongoc_change_stream_next` will block waiting for data'}),
-        ('fullDocument', {'type': 'utf8', 'help': 'A UTF-8 string. Set this option to "updateLookup" to direct the change stream cursor to lookup the most current majority-committed version of the document associated to an update change stream event.'}),
+        ('fullDocument', {
+            'type': 'utf8',
+            'help': 'An optional UTF-8 string. Set this option to "default", '
+                    '"updateLookup", "whenAvailable", or "required", If unset, '
+                    'The string "default" is assumed. Set this option to '
+                    '"updateLookup" to direct the change stream cursor to '
+                    'lookup the most current majority-committed version of the '
+                    'document associated to an update change stream event.'
+        }),
+        ('fullDocumentBeforeChange', {
+            'type': 'utf8',
+            'help': 'An optional UTF-8 string. Set this option to '
+                    '"whenAvailable", "required", or "off". When unset, the '
+                    'default value is "off". Similar to "fullDocument", but '
+                    'returns the value of the document before the associated '
+                    'change.',
+        }),
         comment_option_string_pre_4_4,
-    ], fullDocument="default")),
+    ], fullDocument=None, fullDocumentBeforeChange=None)),
 
     ('mongoc_create_index_opts_t', Struct([
         write_concern_option,

build/opts_templates/mongoc-opts.c.template

@@ -34,7 +34,8 @@ _{{ struct_name }}_parse (
 {% elif info['type'] in ('document', 'array') %}
    bson_init (&{{ struct_name }}->{{ path }});
 {% elif info['type'] == 'utf8' %}
-   {{ struct_name }}->{{ path }} = "{{ description.default(opt_name, "")}}";
+   {% set dflt = description.default(opt_name, "") -%}
+   {{ struct_name }}->{{ path }} = {{ "NULL" if dflt is none else (dflt|tojson) }};
 {% elif info['type'] == 'timestamp' %}
    memset (&{{ struct_name }}->{{ path }}, 0, sizeof (mongoc_timestamp_t));
 {% else %}

src/libmongoc/doc/includes/change-stream-opts.txt

@@ -10,5 +10,6 @@
 * ``startAfter``: A ``Document`` representing the logical starting point of the change stream. Unlike ``resumeAfter``, this can resume notifications after an "invalidate" event. The result of :symbol:`mongoc_change_stream_get_resume_token()` or the ``_id`` field  of any change received from a change stream can be used here.  This option is mutually exclusive with ``resumeAfter`` and ``startAtOperationTime``.
 * ``startAtOperationTime``: A ``Timestamp``. The change stream only provides changes that occurred at or after the specified timestamp. Any command run against the server will return an operation time that can be used here. This option is mutually exclusive with ``resumeAfter`` and ``startAfter``.
 * ``maxAwaitTimeMS``: An ``int64`` representing the maximum amount of time a call to :symbol:`mongoc_change_stream_next` will block waiting for data
-* ``fullDocument``: A UTF-8 string. Set this option to "updateLookup" to direct the change stream cursor to lookup the most current majority-committed version of the document associated to an update change stream event.
+* ``fullDocument``: An optional UTF-8 string. Set this option to "default", "updateLookup", "whenAvailable", or "required", If unset, The string "default" is assumed. Set this option to "updateLookup" to direct the change stream cursor to lookup the most current majority-committed version of the document associated to an update change stream event.
+* ``fullDocumentBeforeChange``: An optional UTF-8 string. Set this option to "whenAvailable", "required", or "off". When unset, the default value is "off". Similar to "fullDocument", but returns the value of the document before the associated change.
 * ``comment``: A :symbol:`bson_value_t` specifying the comment to attach to this command. The comment will appear in log messages, profiler output, and currentOp output. Only string values are supported prior to MongoDB 4.4.

src/libmongoc/src/mongoc/mongoc-change-stream-private.h

@@ -38,6 +38,7 @@ struct _mongoc_change_stream_t {
    bson_t pipeline_to_append;
    bson_t resume_token;
    bson_t *full_document;
+   bson_t *full_document_before_change;
 
    bson_error_t err;
    bson_t err_doc;

src/libmongoc/src/mongoc/mongoc-change-stream.c

@@ -126,6 +126,9 @@ _make_command (mongoc_change_stream_t *stream, bson_t *command)
    if (stream->full_document) {
       bson_concat (&change_stream_doc, stream->full_document);
    }
+   if (stream->full_document_before_change) {
+      bson_concat (&change_stream_doc, stream->full_document_before_change);
+   }
 
    if (stream->resumed) {
       /* Change stream spec: Resume Process */
@@ -418,8 +421,11 @@ _change_stream_init (mongoc_change_stream_t *stream,
    if (stream->opts.fullDocument) {
       stream->full_document =
          BCON_NEW ("fullDocument", stream->opts.fullDocument);
-   } else {
-      stream->full_document = NULL;
+   }
+
+   if (stream->opts.fullDocumentBeforeChange) {
+      stream->full_document_before_change = BCON_NEW (
+         "fullDocumentBeforeChange", stream->opts.fullDocumentBeforeChange);
    }
 
    _mongoc_timestamp_set (&stream->operation_time,
@@ -668,6 +674,7 @@ mongoc_change_stream_destroy (mongoc_change_stream_t *stream)
    bson_destroy (&stream->pipeline_to_append);
    bson_destroy (&stream->resume_token);
    bson_destroy (stream->full_document);
+   bson_destroy (stream->full_document_before_change);
    bson_destroy (&stream->err_doc);
    _mongoc_change_stream_opts_cleanup (&stream->opts);
    mongoc_cursor_destroy (stream->cursor);

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

@@ -145,6 +145,7 @@ typedef struct _mongoc_change_stream_opts_t {
    mongoc_timestamp_t startAtOperationTime;
    int64_t maxAwaitTimeMS;
    const char *fullDocument;
+   const char *fullDocumentBeforeChange;
    bson_value_t comment;
    bson_t extra;
 } mongoc_change_stream_opts_t;

src/libmongoc/src/mongoc/mongoc-opts.c

@@ -1582,6 +1582,7 @@ _mongoc_change_stream_opts_parse (
    memset (&mongoc_change_stream_opts->startAtOperationTime, 0, sizeof (mongoc_timestamp_t));
    mongoc_change_stream_opts->maxAwaitTimeMS = 0;
    mongoc_change_stream_opts->fullDocument = NULL;
+   mongoc_change_stream_opts->fullDocumentBeforeChange = NULL;
    memset (&mongoc_change_stream_opts->comment, 0, sizeof (bson_value_t));
    bson_init (&mongoc_change_stream_opts->extra);
 
@@ -1652,6 +1653,15 @@ _mongoc_change_stream_opts_parse (
             return false;
          }
       }
+      else if (!strcmp (bson_iter_key (&iter), "fullDocumentBeforeChange")) {
+         if (!_mongoc_convert_utf8 (
+               client,
+               &iter,
+               &mongoc_change_stream_opts->fullDocumentBeforeChange,
+               error)) {
+            return false;
+         }
+      }
       else if (!strcmp (bson_iter_key (&iter), "comment")) {
          if (!_mongoc_convert_bson_value_t (
                client,

src/libmongoc/tests/unified/runner.c

@@ -79,8 +79,6 @@ skipped_unified_test_t SKIPPED_TESTS[] = {
    {"cursors are correctly pinned to connections for load-balanced clusters", "listIndexes pins the cursor to a connection"},
    /* libmongoc does not pin connections to cursors. It cannot force an error from waitQueueTimeoutMS by creating cursors in load balanced mode. */
    {"wait queue timeout errors include details about checked out connections", SKIP_ALL_TESTS},
-   /* CDRIVER-4277: Change streams support for user-facing PIT pre- and post-images */
-   {"change-streams-pre_and_post_images", SKIP_ALL_TESTS},
    {0},
 };
 /* clang-format on */