query_stream_result::iterator Class Reference

A single-pass input iterator that synchronously fetches rows one at a time. More...

#include <couchbase/query_stream_result.hxx>

Public Member Functions
auto	operator* () const -> std::pair< error, query_row >
auto	operator++ () -> iterator &
	iterator (std::shared_ptr< internal_query_stream_result > internal)
	Constructs an iterator over an internal result; prefer begin().

Friends
auto	operator== (const iterator &it, end_sentinel) -> bool
	Compares an iterator against the end sentinel.
auto	operator== (end_sentinel end, const iterator &it) -> bool
auto	operator!= (const iterator &it, end_sentinel end) -> bool
auto	operator!= (end_sentinel end, const iterator &it) -> bool

Detailed Description

A single-pass input iterator that synchronously fetches rows one at a time.

Dereferencing yields std::pair<error, query_row>: for a data row the error is falsy; if the stream ends with an error, the iterator visits exactly one final element whose error is truthy (and whose row is empty) before comparing equal to end(). A clean end-of-stream yields no such element. This guarantees a for (auto [err, row] : result) loop can observe a terminal error rather than silently stopping.

#include <couchbase/cluster.hxx>
#include <couchbase/codec/tao_json_serializer.hxx>
 
#include <spdlog/fmt/bundled/chrono.h>
#include <spdlog/fmt/bundled/format.h>
#include <tao/json.hpp>
 
// After the fmt headers: this specializes fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/error.hxx>
 
#include <chrono>
#include <cstddef>
#include <string>
 
int
main(int argc, const char* argv[])
{
  if (argc != 5) {
    fmt::print(
      "USAGE: ./query_stream_iterator couchbase://127.0.0.1 Administrator password default\n");
    return 1;
  }
 
  const std::string connection_string{ argv[1] };
  const std::string username{ argv[2] };
  const std::string password{ argv[3] };
  const std::string bucket_name{ argv[4] };
 
  auto [connect_err, cluster] =
    couchbase::cluster::connect(connection_string, couchbase::cluster_options(username, password))
      .get();
  if (connect_err) {
    fmt::print("unable to connect to the cluster: {}\n", connect_err);
    return 1;
  }
 
  // A scope-level streaming query resolves unqualified keyspaces against that scope, so the
  // statement names the collection (`_default`) rather than the bucket.
  auto scope = cluster.bucket(bucket_name).scope("_default");
 
  { // [1] Range-based for over the stream. Each element is std::pair<error, query_row>: for a data
    //     row the error is falsy. If the stream ends with an error, the iterator visits exactly one
    //     final element carrying that error (with an empty row) before comparing equal to end(), so
    //     the loop can report the failure instead of silently stopping short. A clean end of stream
    //     produces no such element.
    const auto statement = R"(
        SELECT META(d).id AS id, d.name AS name, d.price AS price
        FROM _default AS d
        WHERE d.type = "streaming-example"
        ORDER BY d.price DESC
    )";
    // request_plus, because the documents were written moments ago: with the default
    // (not_bounded) the query runs against whatever the index has caught up with, so a freshly
    // written document may or may not be visible and the row count would vary between runs.
    const auto options =
      couchbase::query_options{}.scan_consistency(couchbase::query_scan_consistency::request_plus);
    auto [err, result] = scope.query_stream(statement, options).get();
    if (err) {
      fmt::print("unable to start the streaming query: {}\n", err);
      return 1;
    }
 
    for (auto [row_err, row] : result) {
      if (row_err) {
        fmt::print("streaming query failed mid-stream: {}\n", row_err);
        return 1;
      }
      const auto p = row.content_as<couchbase::codec::tao_json_serializer, product>();
      fmt::println("{:<24} {:<12} {:>8.2f}", p.id, p.name, p.price);
    }
  }
 
  { // [2] Consuming only a prefix is cheap, and safe. Rows are read from the socket on demand, so
    //     the rows that are never pulled are never transferred and never allocated: the cost of a
    //     prefix read is proportional to the prefix, not to the size of the full result. A consumer
    //     that stops pulling is not penalised either -- the streaming deadline is an inter-read
    //     idle timeout that is armed only while a socket read is in flight, so a slow or partial
    //     consumer produces no socket traffic and is never timed out.
    //
    //     The statement below yields 15000 rows of ~5.2 KB each. Compare the buffered path, which
    //     has to receive and materialise all of them before it resolves, against reading three rows
    //     from the stream and cancelling.
    const auto bulky = R"(SELECT REPEAT("ABCDEFGHIJKLMNOPQRSTUVWXYZ", 200) AS padding
                          FROM ARRAY_RANGE(0, 15000) AS i)";
 
    const auto buffered_started = std::chrono::steady_clock::now();
    auto [buffered_err, buffered] = scope.query(bulky).get();
    const auto buffered_elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
      std::chrono::steady_clock::now() - buffered_started);
    if (buffered_err) {
      fmt::print("unable to perform the buffered query: {}\n", buffered_err);
      return 1;
    }
    std::size_t buffered_bytes{ 0 };
    for (const auto& row : buffered.rows_as_binary()) {
      buffered_bytes += row.size();
    }
    fmt::println("query():        buffered all {} rows ({:.1f} MiB held at once) in {}",
                 buffered.rows_as_binary().size(),
                 static_cast<double>(buffered_bytes) / (1024.0 * 1024.0),
                 buffered_elapsed);
 
    const auto stream_started = std::chrono::steady_clock::now();
    auto [err, result] = scope.query_stream(bulky).get();
    if (err) {
      fmt::print("unable to start the streaming query: {}\n", err);
      return 1;
    }
 
    std::size_t seen{ 0 };
    for (auto [row_err, row] : result) {
      if (row_err) {
        fmt::print("streaming query failed mid-stream: {}\n", row_err);
        return 1;
      }
      if (++seen == 3) {
        break;
      }
    }
    // Release the connection and its timers promptly. Dropping the last handle tears the stream
    // down too, but not until any in-flight pull settles.
    result.cancel();
    const auto stream_elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
      std::chrono::steady_clock::now() - stream_started);
    fmt::println("query_stream(): read {} rows and cancelled in {} -- the rest was never fetched",
                 seen,
                 stream_elapsed);
    // The streaming figure covers the same statement, run second, so it is if anything pessimistic
    // with respect to server-side caching.
  }
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./query_stream_iterator couchbase://127.0.0.1 Administrator password default
streaming-example-9      widget-9        19.00
streaming-example-8      widget-8        18.00
streaming-example-7      widget-7        17.00
streaming-example-6      widget-6        16.00
streaming-example-5      widget-5        15.00
streaming-example-4      widget-4        14.00
streaming-example-3      widget-3        13.00
streaming-example-2      widget-2        12.00
streaming-example-1      widget-1        11.00
streaming-example-0      widget-0        10.00
query():        buffered all 15000 rows (74.6 MiB held at once) in 621ms
query_stream(): read 3 rows and cancelled in 12ms -- the rest was never fetched
 
 */

Part [2] of the error-handling example shows the terminal element being observed rather than silently truncating the result:

#include <couchbase/cluster.hxx>
#include <couchbase/error_codes.hxx>
 
#include <spdlog/fmt/bundled/format.h>
 
// After the fmt header: this specializes fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/error.hxx>
 
#include <chrono>
#include <cstdint>
#include <string>
 
namespace
{
// How far a drain got. A streaming query is not all-or-nothing the way a buffered query() is: it
// can hand over rows and only then fail, and those rows were valid. An application that mutates
// state per row has to decide whether that partial work is acceptable, so the row count travels
// with the error rather than being discarded.
struct drain_report {
  couchbase::error error{};
  std::uint64_t rows{ 0 };
};
 
auto
drain(const couchbase::query_stream_result& result) -> drain_report
{
  drain_report report{};
  while (true) {
    auto [row_err, row] = result.next().get();
    if (row_err) {
      report.error = std::move(row_err); // terminal error: no further row will arrive
      return report;
    }
    if (!row) {
      return report; // clean end of stream
    }
    ++report.rows;
  }
}
} // namespace
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./query_stream_errors couchbase://127.0.0.1 Administrator password\n");
    return 1;
  }
 
  const std::string connection_string{ argv[1] };
  const std::string username{ argv[2] };
  const std::string password{ argv[3] };
 
  auto [connect_err, cluster] =
    couchbase::cluster::connect(connection_string, couchbase::cluster_options(username, password))
      .get();
  if (connect_err) {
    fmt::print("unable to connect to the cluster: {}\n", connect_err);
    return 1;
  }
 
  // A syntactically valid statement over a keyspace that does not exist. Where the failure
  // surfaces is not fixed: the query service may reject the request while the SDK is still
  // reading the response preamble, or accept it and then terminate the stream. Both channels
  // have to be handled -- a caller that only checks one of them will miss the failure.
  const auto invalid_statement = R"(SELECT * FROM `nonexistent_keyspace_xyz` LIMIT 1)";
  bool saw_error{ false };
 
  { // [1] Both channels for one failing statement. The error is either carried by the
    //     query_stream() future (the request never started, and the returned handle is not a
    //     usable stream), or delivered as the stream's terminal by next(). The row count that
    //     travels with the terminal is the part of the result that was consumed successfully.
    // Formatting an error renders its context too: the code and message the service itself
    // reported, the statement and encoded parameters, and the node that answered. That is what to
    // log. couchbase::error::ctx().to_json() returns the same context on its own, for a structured
    // log line.
    auto [err, result] = cluster.query_stream(invalid_statement).get();
    if (err) {
      saw_error = true;
      fmt::println("[1] rejected before streaming began: {}", err);
    } else {
      auto report = drain(result);
      if (report.error) {
        saw_error = true;
        fmt::println("[1] stream failed after {} row(s): {}", report.rows, report.error);
      } else {
        fmt::println("[1] stream ended cleanly with {} row(s)", report.rows);
      }
    }
  }
 
  { // [2] The iterator reports a terminal error as one final element (truthy error, empty row)
    //     before it compares equal to end(), so a range-based for loop surfaces the failure
    //     instead of quietly stopping short of the result. Ignoring the error half of the pair is
    //     how a truncated result gets mistaken for a complete one.
    auto [err, result] = cluster.query_stream(invalid_statement).get();
    if (err) {
      saw_error = true;
      fmt::println("[2] rejected before streaming began: {}", err);
    } else {
      std::uint64_t rows{ 0 };
      for (auto [row_err, row] : result) {
        if (row_err) {
          saw_error = true;
          fmt::println("[2] iterator saw the terminal error after {} row(s): {}", rows, row_err);
          break;
        }
        ++rows;
      }
    }
  }
 
  { // [3] Deliberate teardown is a terminal too, and has to be told apart from a real failure:
    //     after cancel(), next() reports errc::common::request_canceled. The rows consumed before
    //     the cancel are unaffected -- this is the partial-success case that a buffered query()
    //     cannot produce.
    auto [err, result] = cluster.query_stream("SELECT n FROM ARRAY_RANGE(0, 15000) AS n").get();
    if (err) {
      fmt::print("unable to start the streaming query: {}\n", err);
      return 1;
    }
 
    std::uint64_t rows{ 0 };
    couchbase::error before_cancel{};
    while (rows < 3) {
      auto [row_err, row] = result.next().get();
      if (row_err) {
        before_cancel = std::move(row_err);
        break;
      }
      if (!row) {
        break; // the result was shorter than expected
      }
      ++rows;
    }
    result.cancel();
 
    const auto after_cancel = drain(result);
    if (before_cancel) {
      fmt::println("[3] the stream failed before it could be cancelled: {}", before_cancel);
    } else if (after_cancel.error.ec() == couchbase::errc::common::request_canceled) {
      fmt::println("[3] stream cancelled after {} row(s); those rows are still valid", rows);
    } else {
      fmt::println("[3] unexpected terminal after cancel(): \"{}\"",
                   after_cancel.error.ec().message());
    }
 
    // [4] The terminal is sticky and idempotent: further next() calls re-deliver it rather than
    //     blocking on a drained stream, so a drain loop can never hang on a dead stream.
    auto [again_err, again_row] = result.next().get();
    fmt::println("[4] re-reading past the terminal: error=\"{}\", row_present={}",
                 again_err.ec().message(),
                 again_row.has_value());
 
    // [5] meta_data() resolves with the failure instead of parking forever waiting for a trailer
    //     that will never arrive.
    auto [meta_err, meta] = result.meta_data().get();
    fmt::println("[5] meta_data() after a torn-down stream: \"{}\"", meta_err.ec().message());
  }
 
  { // [6] Timeouts are classified by whether the request could have applied a mutation:
    //     errc::common::unambiguous_timeout when the request is read-only (it definitely did not
    //     apply, so retrying is safe) and errc::common::ambiguous_timeout otherwise (it may
    //     already have been applied -- do not blindly retry). The classification follows the
    //     request's read-only flag, which defaults to false: the SDK does not infer read-only-ness
    //     from the statement text, so mark read-only queries explicitly to get the retryable
    //     classification.
    //
    //     The deadline applies as a whole-request timeout until the response headers arrive, and
    //     from then on as an *inter-read idle* timeout that is armed only while a socket read is in
    //     flight. So a slow consumer, which generates no socket traffic, is never timed out; a
    //     mid-stream fire means the server stalled mid-body. Either way the classification is the
    //     same, and the code below handles the deadline firing on either side of the preamble.
    const auto options = couchbase::query_options{}.readonly(true).timeout(
      std::chrono::milliseconds{ 1 }); // far too short, on purpose
    auto [err, result] =
      cluster.query_stream("SELECT n FROM ARRAY_RANGE(0, 15000) AS n", options).get();
    auto ec = err.ec();
    std::uint64_t rows{ 0 };
    if (!ec) {
      const auto report = drain(result);
      ec = report.error.ec();
      rows = report.rows;
    }
    if (ec == couchbase::errc::common::unambiguous_timeout) {
      fmt::println("[6] read-only statement timed out after {} row(s); retrying it is safe", rows);
    } else if (ec == couchbase::errc::common::ambiguous_timeout) {
      fmt::println("[6] statement timed out ambiguously after {} row(s); do not blindly retry",
                   rows);
    } else {
      fmt::println("[6] statement did not time out ({} row(s), \"{}\")", rows, ec.message());
    }
  }
 
  if (!saw_error) {
    fmt::print("expected the invalid statement to fail, but it did not\n");
    return 1;
  }
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./query_stream_errors couchbase://127.0.0.1 Administrator password
[1] rejected before streaming began: index_failure (202) | {"client_context_id":"1f2170-81d8-224f-
    45e4-6df76260782b22","first_error_code":12003,"first_error_message":"Keyspace not found in CB
    datastore: default:nonexistent_keyspace_xyz (near line 1, column 15) - cause: No bucket named
    nonexistent_keyspace_xyz","hostname":"172.18.0.5", ... ,"statement":"SELECT * FROM
    `nonexistent_keyspace_xyz` LIMIT 1"}
    (one line in reality; wrapped here, and the remaining context fields elided, for readability)
[2] rejected before streaming began: index_failure (202) | { ... same context ... }
[3] stream cancelled after 3 row(s); those rows are still valid
[4] re-reading past the terminal: error="request_canceled (2)", row_present=false
[5] meta_data() after a torn-down stream: "request_canceled (2)"
[6] read-only statement timed out after 0 row(s); retrying it is safe
 
 */

This is a minimal single-pass iterator intended for range-based for and manual while (it != result.end()) loops. It compares only against end() (never against another iterator), and provides neither operator-> nor post-increment, so it is deliberately not a LegacyInputIterator and does not work with <algorithm> or std::distance. begin() must be called once.

Since

1.4.0

Volatile

Should not be used in production

Constructor & Destructor Documentation

iterator()

iterator ( std::shared_ptr< internal_query_stream_result > internal )

explicit

Constructs an iterator over an internal result; prefer begin().

Since

1.4.0

Internal

Internal interface

Member Function Documentation

operator*()

auto operator*

(

)

const -> std::pair< error, query_row >

operator++()

auto operator++

(

)

-> iterator &

Friends And Related Symbol Documentation

operator!= [1/2]

auto operator!= ( const iterator & it, end_sentinel end ) -> bool

friend

operator!= [2/2]

auto operator!= ( end_sentinel end, const iterator & it ) -> bool

friend

operator== [1/2]

auto operator== ( const iterator & it, end_sentinel  ) -> bool

friend

Compares an iterator against the end sentinel.

Provided in both argument orders (and as both == and !=) so the range-for idiom, while (it != result.end()) loops, and wrapper code all work regardless of which operand is written first. Iterators are intentionally never comparable to one another: a single-pass stream has one position, so iterator-to-iterator equality would be meaningless — the type only ever compares against end().

operator== [2/2]

auto operator== ( end_sentinel end, const iterator & it ) -> bool

friend

---

The documentation for this class was generated from the following file:

• couchbase/query_stream_result.hxx