scope Class Reference

The scope identifies a group of collections and allows high application density as a result. More...

#include <couchbase/scope.hxx>

Public Member Functions
auto	bucket_name () const -> const std::string &
	Returns name of the bucket where the scope is defined.
auto	name () const -> const std::string &
	Returns name of the scope.
auto	collection (std::string_view collection_name) const -> collection
	Opens a collection for this scope with an explicit name.
void	query (std::string statement, const query_options &options, query_handler &&handler) const
	Performs a query against the query (N1QL) services.
auto	query (std::string statement, const query_options &options={}) const -> std::future< std::pair< error, query_result > >
	Performs a query against the query (N1QL) services.
void	query_stream (std::string statement, const query_options &options, query_stream_handler &&handler) const
	Performs a streaming query against the query (N1QL) services.
auto	query_stream (std::string statement, const query_options &options={}) const -> std::future< std::pair< error, query_stream_result > >
	Performs a streaming query against the query (N1QL) services.
void	search (std::string index_name, search_request request, const search_options &options, search_handler &&handler) const
	Performs a request against the full text search services.
auto	search (std::string index_name, search_request request, const search_options &options={}) const -> std::future< std::pair< error, search_result > >
	Performs a request against the full text search services.
void	analytics_query (std::string statement, const analytics_options &options, analytics_handler &&handler) const
	Performs a query against the analytics services.
auto	analytics_query (std::string statement, const analytics_options &options={}) const -> std::future< std::pair< error, analytics_result > >
	Performs a query against the analytics services.
void	analytics_query_stream (std::string statement, const analytics_options &options, analytics_stream_handler &&handler) const
	Performs a streaming query against the analytics services.
auto	analytics_query_stream (std::string statement, const analytics_options &options={}) const -> std::future< std::pair< error, analytics_stream_result > >
	Performs a streaming query against the analytics services.
auto	search_indexes () const -> scope_search_index_manager
	Provides access to search index management services at the scope level.

Static Public Attributes
static constexpr auto	default_name { "_default" }
	Constant for the name of the default scope in the bucket.

Friends
class	bucket

Detailed Description

The scope identifies a group of collections and allows high application density as a result.

Since

1.0.0

Member Function Documentation

analytics_query() [1/2]

void analytics_query	(	std::string	statement,
		const analytics_options &	options,
		analytics_handler &&	handler ) const

Performs a query against the analytics services.

Parameters

statement	the query statement.
options	options to customize the query request.
handler	the handler that implements query_handler

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

See also

https://docs.couchbase.com/server/current/analytics/introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

analytics_query() [2/2]

auto analytics_query ( std::string statement, const analytics_options & options = {} ) const -> std::future< std::pair< error, analytics_result > >

nodiscard

Performs a query against the analytics services.

Parameters

statement	the query statement.
options	options to customize the query request.

Returns

future object that carries result of the operation

See also

https://docs.couchbase.com/server/current/analytics/introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

analytics_query_stream() [1/2]

void analytics_query_stream	(	std::string	statement,
		const analytics_options &	options,
		analytics_stream_handler &&	handler ) const

Performs a streaming query against the analytics services.

The handler resolves as soon as the response preamble has been parsed; rows are then pulled lazily from the analytics_stream_result, so the full result is never buffered in memory.

Parameters

statement	the analytics query statement.
options	options to customize the query request.
handler	the handler that implements analytics_stream_handler

Since

1.4.0

Volatile

Should not be used in production

analytics_query_stream() [2/2]

auto analytics_query_stream ( std::string statement, const analytics_options & options = {} ) const -> std::future< std::pair< error, analytics_stream_result > >

nodiscard

Performs a streaming query against the analytics services.

The example is written against analytics_query_stream(), but the scope-level overload behaves identically apart from resolving the statement against this scope:

#include <couchbase/cluster.hxx>
#include <couchbase/codec/tao_json_serializer.hxx>
 
#include <spdlog/fmt/bundled/format.h>
#include <tao/json.hpp>
 
// After the fmt header: these specialize fmt::formatter, so fmt has to be declared first.
#include <couchbase/fmt/analytics_status.hxx>
#include <couchbase/fmt/error.hxx>
 
#include <cstdint>
#include <string>
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./analytics_stream 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;
  }
 
  // Analytics streaming mirrors the query API: the same three-state next(), the same iterator,
  // the same end-of-stream metadata.
  auto [err, result] =
    cluster.analytics_query_stream("SELECT i AS n FROM array_range(0, 2000) AS i ORDER BY i").get();
  if (err) {
    fmt::print("unable to start the streaming analytics query: {}\n", err);
    return 1;
  }
 
  // The row signature comes from the response preamble, so it is available before the stream has
  // been drained (unlike the metadata).
  if (const auto signature = result.signature(); signature) {
    // codec::binary is a byte vector; reinterpret to chars to print it as the JSON text it is.
    fmt::println(
      "signature: {}",
      std::string{ reinterpret_cast<const char*>(signature->data()), signature->size() });
  }
 
  std::uint64_t sum{ 0 };
  while (true) {
    auto [row_err, row] = result.next().get();
    if (row_err) {
      fmt::print("streaming analytics query failed mid-stream: {}\n", row_err);
      return 1;
    }
    if (!row) {
      break; // clean end of stream
    }
    const auto value = row->content_as<couchbase::codec::tao_json_serializer, tao::json::value>();
    sum += value.at("n").as<std::uint64_t>();
  }
  fmt::println("sum of the streamed rows: {}", sum);
 
  auto [meta_err, meta] = result.meta_data().get();
  if (meta_err) {
    fmt::print("unable to retrieve the analytics metadata: {}\n", meta_err);
    return 1;
  }
  fmt::println("status={}, rows={}", meta.status(), meta.metrics().result_count());
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./analytics_stream couchbase://127.0.0.1 Administrator password
signature: {"*":"*"}
sum of the streamed rows: 1999000
status=success, rows=2000
 
 */

Parameters

statement	the analytics query statement.
options	options to customize the query request.

Returns

future object that carries the streaming result handle

Since

1.4.0

Volatile

Should not be used in production

bucket_name()

auto bucket_name ( ) const -> const std::string &

nodiscard

Returns name of the bucket where the scope is defined.

Returns

name of the bucket

Since

1.0.0

Committed

Generally available API and should be preferred in production

collection()

auto collection ( std::string_view collection_name ) const -> collection

nodiscard

Opens a collection for this scope with an explicit name.

Parameters

collection_name

the collection name.

Returns

the requested collection if successful.

Since

1.0.0

Committed

Generally available API and should be preferred in production

name()

auto name ( ) const -> const std::string &

nodiscard

Returns name of the scope.

Returns

name of the scope

Since

1.0.0

Committed

Generally available API and should be preferred in production

query() [1/2]

void query	(	std::string	statement,
		const query_options &	options,
		query_handler &&	handler ) const

Performs a query against the query (N1QL) services.

Parameters

statement	the N1QL query statement.
options	options to customize the query request.
handler	the handler that implements query_handler

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

Since

1.0.0

Committed

Generally available API and should be preferred in production

query() [2/2]

auto query ( std::string statement, const query_options & options = {} ) const -> std::future< std::pair< error, query_result > >

nodiscard

Performs a query against the query (N1QL) services.

Parameters

statement	the N1QL query statement.
options	options to customize the query request.

Returns

future object that carries result of the operation

Since

1.0.0

Committed

Generally available API and should be preferred in production

query_stream() [1/2]

void query_stream	(	std::string	statement,
		const query_options &	options,
		query_stream_handler &&	handler ) const

Performs a streaming query against the query (N1QL) services.

The handler resolves as soon as the response preamble has been parsed; rows are then pulled lazily from the query_stream_result, so the full result is never buffered in memory.

Note

Prepared statements (adhoc set to false) are not streamed: such a request falls back to the buffered query() path and its rows are replayed through the returned handle.

The handler runs on the library's I/O thread, so the stream must be drained with the callback next() overload rather than the blocking ones. The example is written against query_stream(), but the scope-level overload behaves identically:

#include <couchbase/cluster.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 <cstdint>
#include <future>
#include <memory>
#include <optional>
#include <string>
#include <utility>
 
// A drain that never blocks a thread: each next() completion issues the following pull from
// inside the completion handler, so the whole result is consumed on the library's I/O threads.
//
// The blocking overloads -- next().get(), meta_data().get(), and the eager iterator -- must NOT
// be used from a completion handler: they park the very thread that has to deliver the row, and
// the stream can never advance. From a handler, always use the callback next() overload.
//
// Chaining the next pull from inside the handler does not grow the stack: a row is delivered from
// the I/O event loop rather than synchronously from within next(), so the pulls do not nest.
class row_counter : public std::enable_shared_from_this<row_counter>
{
public:
  using result_type = std::pair<couchbase::error, std::uint64_t>;
 
  row_counter(couchbase::query_stream_result result,
              std::shared_ptr<std::promise<result_type>> done)
    : result_{ std::move(result) }
    , done_{ std::move(done) }
  {
  }
 
  void pull()
  {
    // shared_from_this() keeps the counter (and the stream handle it owns) alive for exactly as
    // long as pulls are outstanding.
    result_.next(
      [self = shared_from_this()](couchbase::error err, std::optional<couchbase::query_row> row) {
        if (err) {
          self->done_->set_value({ std::move(err), self->rows_ });
          return;
        }
        if (!row) {
          self->done_->set_value({ {}, self->rows_ }); // clean end of stream
          return;
        }
        ++self->rows_;
        self->pull();
      });
  }
 
private:
  couchbase::query_stream_result result_;
  std::shared_ptr<std::promise<result_type>> done_;
  std::uint64_t rows_{ 0 };
};
 
int
main(int argc, const char* argv[])
{
  if (argc != 4) {
    fmt::print("USAGE: ./query_stream_async 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;
  }
 
  // However many rows the statement returns, the SDK holds only a bounded window of them: it stops
  // reading the socket once the rows it has buffered exceed a high-water mark and resumes once the
  // consumer drains back below the low-water mark. The consumer sets the pace.
  auto done = std::make_shared<std::promise<row_counter::result_type>>();
  auto finished = done->get_future();
 
  cluster.query_stream("SELECT n FROM ARRAY_RANGE(0, 5000) AS n",
                       couchbase::query_options{},
                       [done](couchbase::error err, couchbase::query_stream_result result) {
                         if (err) {
                           done->set_value({ std::move(err), 0 });
                           return;
                         }
                         std::make_shared<row_counter>(std::move(result), done)->pull();
                       });
 
  // The pump reports the row count alongside the error, because a stream can fail *after* rows
  // have already been delivered: those rows were valid and any work done on them stands. See the
  // error-handling example for the full set of failure channels.
  auto [stream_err, rows] = finished.get();
  if (stream_err) {
    fmt::print("streaming query failed after {} row(s): {}\n", rows, stream_err);
    return 1;
  }
  fmt::println("counted {} rows without blocking a thread", rows);
 
  cluster.close().get();
  return 0;
}
 
/*
 
$ ./query_stream_async couchbase://127.0.0.1 Administrator password
counted 5000 rows without blocking a thread
 
 */

Parameters

statement	the N1QL query statement.
options	options to customize the query request.
handler	the handler that implements query_stream_handler

Since

1.4.0

Volatile

Should not be used in production

query_stream() [2/2]

auto query_stream ( std::string statement, const query_options & options = {} ) const -> std::future< std::pair< error, query_stream_result > >

nodiscard

Performs a streaming query against the query (N1QL) services.

Unqualified keyspaces in the statement resolve against this scope, so it names the collection rather than the bucket. The result can be consumed with a range-based for loop, and abandoned early without transferring the rows that were never read:

#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
 
 */

Parameters

statement	the N1QL query statement.
options	options to customize the query request.

Returns

future object that carries the streaming result handle

Since

1.4.0

Volatile

Should not be used in production

search() [1/2]

void search	(	std::string	index_name,
		search_request	request,
		const search_options &	options,
		search_handler &&	handler ) const

Performs a request against the full text search services.

This can be used to perform a traditional FTS query, and/or a vector search.

Parameters

index_name	name of the search index
request	request object, see search_request for more details.
options	options to customize the query request.
handler	the handler that implements search_handler

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

See also

https://docs.couchbase.com/server/current/fts/fts-introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

search() [2/2]

auto search ( std::string index_name, search_request request, const search_options & options = {} ) const -> std::future< std::pair< error, search_result > >

nodiscard

Performs a request against the full text search services.

This can be used to perform a traditional FTS query, and/or a vector search.

Parameters

index_name	name of the search index
request	request object, see search_request for more details.
options	options to customize the query request.

Returns

future object that carries result of the operation

Exceptions

errc::common::ambiguous_timeout
errc::common::unambiguous_timeout

See also

https://docs.couchbase.com/server/current/fts/fts-introduction.html

Since

1.0.0

Committed

Generally available API and should be preferred in production

search_indexes()

auto search_indexes ( ) const -> scope_search_index_manager

nodiscard

Provides access to search index management services at the scope level.

Returns

a manager instance

Since

1.0.0

Committed

Generally available API and should be preferred in production

Friends And Related Symbol Documentation

bucket

friend class bucket

friend

Member Data Documentation

default_name

auto default_name { "_default" }

staticconstexpr

Constant for the name of the default scope in the bucket.

Since

1.0.0

Committed

Generally available API and should be preferred in production

Examples

distributed_mutex.cxx, and minimal.cxx.

---

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

• couchbase/scope.hxx