Scan Consistency Using the C SDK with Couchbase Server

You can balance performance against consistency in N1QL queries via the LCB client and the AtPlus option.

Using CONSISTENCY_REQUEST guarantees that the query will include newly-inserted documents:

#include <libcouchbase/couchbase.h>
#include <libcouchbase/api3.h>
#include <libcouchbase/n1ql.h>
#include <string>
#include <vector>

static int RandomNumber_g;
typedef std::vector<std::string> RowList;

static void query_callback(lcb_t, int, const lcb_RESPN1QL *resp)
{
    if (resp->rc != LCB_SUCCESS) {
        fprintf(stderr, "N1QL query failed (%s)\n", lcb_strerror(NULL, resp->rc));
    }

    if (resp->rflags & LCB_RESP_F_FINAL) {
        // We're simply notified here that the last row has already been returned.
        // no processing needed here.
        return;
    }

    // Add rows to the vector, we'll process the results in the calling
    // code.
    RowList *rowlist = reinterpret_cast<RowList*>(resp->cookie);
    rowlist->push_back(std::string(resp->row, resp->nrow));
}

int
main(int, char **)
{
    lcb_t instance;
    lcb_create_st crst;
    lcb_error_t rc;

    memset(&crst, 0, sizeof crst);
    crst.version = 3;
    crst.v.v3.connstr = "couchbase://10.0.0.31/default";

    rc = lcb_create(&instance, &crst);
    rc = lcb_connect(instance);
    lcb_wait(instance);
    rc = lcb_get_bootstrap_status(instance);

    // Initialize random seed to get a "random" value in our documents
    srand(time(NULL));
    RandomNumber_g = rand() % 10000;

    char key[256];
    sprintf(key, "user:%d", RandomNumber_g);
    char value[4096];
    sprintf(value,
        "{"
            "\"name\":[\"Brass\",\"Doorknob\"],"
            "\"email\":\"brass.doorknob@juno.com\","
            "\"random\":%d"
        "}", RandomNumber_g);

    printf("Will insert new document with random number %d\n", RandomNumber_g);

    lcb_CMDSTORE scmd = { 0 };
    LCB_CMD_SET_KEY(&scmd, key, strlen(key));
    LCB_CMD_SET_VALUE(&scmd, value, strlen(value));
    scmd.operation = LCB_SET;

    lcb_sched_enter(instance);
    rc = lcb_store3(instance, NULL, &scmd);
    lcb_sched_leave(instance);
    lcb_wait(instance);
    // In real code, we'd also install a store callback so we can know if the
    // actual storage operation was a success.

    lcb_N1QLPARAMS *params = lcb_n1p_new();

    lcb_CMDN1QL cmd = { 0 };
    RowList rows;
    cmd.callback = query_callback;

    rc = lcb_n1p_setstmtz(params,
        "SELECT name, email, random FROM default WHERE $1 IN name");
    // -1 for length indicates nul-terminated string
    rc = lcb_n1p_posparam(params, "\"Brass\"", -1);

    // This guarantess that the query will include the newly-inserted document.
    rc = lcb_n1p_setconsistency(params, LCB_N1P_CONSISTENCY_REQUEST);
    rc = lcb_n1p_mkcmd(params, &cmd);
    rc = lcb_n1ql_query(instance, &rows, &cmd);
    lcb_wait(instance);

    // To demonstrate the CONSISTENCY_REQUEST feature, we check each row for the
    // "random" value. Because the C standard library does not come with a JSON
    // parser, we are limited in how we can inspect the row (which is JSON).
    // For clarity, we print out only the row's "Random" field. When the
    // CONSISTENCY_REQUEST feature is enabled, one of the results should contain
    // the newest random number (the value of RandomNumber_g). When disabled, the
    // row may or may not appear.
    for (RowList::iterator ii = rows.begin(); ii != rows.end(); ++ii) {
        std::string& row = *ii;
        size_t begin_pos = row.find("\"random\"");
        size_t end_pos = row.find_first_of("},", begin_pos);
        std::string row_number = row.substr(begin_pos, end_pos - begin_pos);
        printf("Row has random number %s\n", row_number.c_str());
    }

    lcb_n1p_free(params);
    lcb_destroy(instance);
}

Setting a Scan Consistency, lets you choose between NotBounded (the default), for sharpest performance; RequestPlus, for strongest consistency; and AtPlus, for a good balance between increased performance and completeness of results:>

#include <libcouchbase/couchbase.h>
#include <libcouchbase/n1ql.h>
#include <string>
#include <vector>
#include <sstream>

static int RandomNumber_g;
typedef std::vector<std::string> RowList;

static void query_callback(lcb_t, int, const lcb_RESPN1QL *resp)
{
    if (resp->rc != LCB_SUCCESS) {
        fprintf(stderr, "N1QL query failed (%s)\n", lcb_strerror(NULL, resp->rc));
    }

    if (resp->rflags & LCB_RESP_F_FINAL) {
        // We're simply notified here that the last row has already been returned.
        // no processing needed here.
        return;
    }

    // Add rows to the vector, we'll process the results in the calling
    // code.
    RowList *rowlist = reinterpret_cast<RowList*>(resp->cookie);
    rowlist->push_back(std::string(resp->row, resp->nrow));
}

static void storage_callback(lcb_t, int cbtype, const lcb_RESPBASE *resp)
{
    if (resp->rc != LCB_SUCCESS) {
        fprintf(stderr, "Failed to store document: %s\n", lcb_strerror(NULL, resp->rc));
        exit(EXIT_FAILURE);
    }

    lcb_MUTATION_TOKEN *mt = reinterpret_cast<lcb_MUTATION_TOKEN*>(resp->cookie);
    *mt = *lcb_resp_get_mutation_token(cbtype, resp);
}

int
main(int, char **)
{
    lcb_t instance;
    lcb_create_st crst;
    lcb_error_t rc;

    memset(&crst, 0, sizeof crst);
    crst.version = 3;
    crst.v.v3.connstr = "couchbase://localhost/default?fetch_mutation_tokens=true";

    rc = lcb_create(&instance, &crst);
    rc = lcb_connect(instance);
    lcb_wait(instance);
    rc = lcb_get_bootstrap_status(instance);

    // Initialize random seed to get a "random" value in our documents
    srand(time(NULL));
    RandomNumber_g = rand() % 10000;

    // Install the storage callback which will be used to retrieve the
    // mutation token
    lcb_install_callback3(instance, LCB_CALLBACK_STORE, storage_callback);

    lcb_MUTATION_TOKEN mt;
    memset(&mt, 0, sizeof mt);

    char key[256];
    sprintf(key, "user:%d", RandomNumber_g);
    char value[4096];
    sprintf(value,
        "{"
            "\"name\":[\"Brass\",\"Doorknob\"],"
            "\"email\":\"brass.doorknob@juno.com\","
            "\"random\":%d"
        "}", RandomNumber_g);

    printf("Will insert new document with random number %d\n", RandomNumber_g);

    lcb_CMDSTORE scmd = { 0 };
    LCB_CMD_SET_KEY(&scmd, key, strlen(key));
    LCB_CMD_SET_VALUE(&scmd, value, strlen(value));

    rc = lcb_store3(instance, &mt, &scmd);
    lcb_wait(instance);

    lcb_CMDN1QL cmd = { 0 };
    RowList rows;
    cmd.callback = query_callback;

    // At the time of writing, the lcb_N1QLPARAMS implementation has some
    // bugs in it with respect to adding mutation tokens. For this reason, we're
    // encoding the query manually. This would look a bit nicer if we were
    // using a real JSON library:

    const char *bucketname;
    lcb_cntl(instance, LCB_CNTL_GET, LCB_CNTL_BUCKETNAME, &bucketname);
    std::stringstream stmt;

    stmt <<
            "{"
            "\"statement\":\"SELECT name, email, random FROM default WHERE $1 in name\","
            "\"args\":[\"Brass\"],"
            "\"scan_consistency\":\"at_plus\",";

    stmt << "\"scan_vectors\":{"
            << "\"" <<  bucketname << "\":{"
                << "\"" <  LCB_MUTATION_TOKEN_VB(&mt) < "\":["
                    << LCB_MUTATION_TOKEN_SEQ(&mt) << ","
                    << "\"" < LCB_MUTATION_TOKEN_ID(&mt) << "\""
                << "]"
             << "}"
           << "}"
         << "}";

    /*
    The above expands to something like this:
    {
        "statement": "SELECT name, email, random FROM default WHERE $1 in name",
        "args": ["Brass"],
        "scan_consistency": "at_plus",
        "scan_vectors": {
            "default": {
                "29": [3, "88598346863273"]
            }
        }
    }
    */

    std::string stmt_str = stmt.str();
    cmd.query = stmt_str.c_str();
    cmd.nquery = stmt_str.size();
    rc = lcb_n1ql_query(instance, &rows, &cmd);
    assert(rc==LCB_SUCCESS);
    lcb_wait(instance);

    // To demonstrate the at_plus feature,
    // we check each row for the "random" value.
    // Because the C standard library does not come with a JSON
    // parser, we are limited in how we can inspect the row (which is JSON).
    // For clarity, we print out only the row's "Random" field. When the
    // at_plus feature is enabled, one of the results should contain
    // the newest random number (the value of RandomNumber_g). When disabled, the
    // row may or may not appear.
    for (RowList::iterator ii = rows.begin(); ii != rows.end(); ++ii) {
        std::string& row = *ii;
        size_t begin_pos = row.find("\"random\"");
        size_t end_pos = row.find_first_of("},", begin_pos);
        std::string row_number = row.substr(begin_pos, end_pos - begin_pos);
        printf("Row has random number %s\n", row_number.c_str());
    }

    lcb_destroy(instance);
}