ustore_database_t db_ = nullptr;

ustore_collection_t collection_ = ustore_collection_main_k;

ustore_transaction_t txn_ = nullptr;

ustore_snapshot_t snap_ = 0;

arena_t arena_scan_;

arena_t arena_read_;

ustore_length_t read_ahead_ = 0;

ustore_key_t next_min_key_ = std::numeric_limits<ustore_key_t>::min();

ptr_range_gt<ustore_key_t> fetched_keys_;

embedded_blobs_t values_view;

std::size_t fetched_offset_ = 0;

status_t prefetch() noexcept {

if (next_min_key_ == ustore_key_unknown_k)

return {};

ustore_key_t* found_keys {};

ustore_length_t* found_offsets {};

ustore_length_t* found_counts {};

ustore_length_t* found_lengths {};

ustore_bytes_ptr_t found_values {};

ustore_str_view_t fields {};

status_t status {};

ustore_scan_t scan {};

scan.db = db_;

scan.error = status.member_ptr();

scan.transaction = txn_;

scan.arena = arena_scan_.member_ptr();

scan.tasks_count = 1;

scan.collections = &collection_;

scan.start_keys = &next_min_key_;

scan.count_limits = &read_ahead_;

scan.offsets = &found_offsets;

scan.counts = &found_counts;

scan.keys = &found_keys;

ustore_scan(&scan);

if (!status)

return status;

fetched_keys_ = ptr_range_gt<ustore_key_t> {found_keys, found_keys + *found_counts};

fetched_offset_ = 0;

auto count = static_cast<ustore_size_t>(fetched_keys_.size());

ustore_docs_read_t docs_read {};

docs_read.db = db_;

docs_read.error = status.member_ptr();

docs_read.transaction = txn_;

docs_read.snapshot = snap_;

docs_read.arena = arena_read_.member_ptr();

docs_read.type = ustore_doc_field_json_k;

docs_read.tasks_count = count;

docs_read.collections = &collection_;

docs_read.keys = found_keys;

docs_read.keys_stride = sizeof(ustore_key_t);

docs_read.fields = &fields;

docs_read.fields_stride = 0;

docs_read.offsets = &found_offsets;

docs_read.lengths = &found_lengths;

docs_read.values = &found_values;

ustore_docs_read(&docs_read);

if (!status)

return status;

values_view = embedded_blobs_t {count, found_offsets, found_lengths, found_values};

next_min_key_ = count <= read_ahead_ ? ustore_key_unknown_k : fetched_keys_[count - 1] + 1;

return {};

}

public:

static constexpr std::size_t default_read_ahead_k = 256;

docs_pairs_stream_t(ustore_database_t db,

ustore_collection_t collection = ustore_collection_main_k,

std::size_t read_ahead = docs_pairs_stream_t::default_read_ahead_k,

ustore_transaction_t txn = nullptr,

ustore_snapshot_t snap = {})

: db_(db), collection_(collection), txn_(txn), snap_(snap), arena_scan_(db_), arena_read_(db_),

read_ahead_(static_cast<ustore_size_t>(read_ahead)) {}

status_t seek(ustore_key_t key) noexcept {

fetched_keys_ = {};

fetched_offset_ = 0;

next_min_key_ = key;

return prefetch();

}

status_t advance() noexcept {

if (fetched_offset_ >= fetched_keys_.size())

return prefetch();

++fetched_offset_;

return {};

}

docs_pairs_stream_t& operator++() noexcept {

status_t status = advance();

if (status)

return *this;

fetched_keys_ = {};

fetched_offset_ = 0;

next_min_key_ = ustore_key_unknown_k;

return *this;

}

ustore_key_t key() const noexcept { return fetched_keys_[fetched_offset_]; }

value_view_t value() const noexcept {

auto it = values_view.begin();

for (size_t i = 0; i != fetched_offset_; ++i)

++it;

return *it;

}

bool is_end() const noexcept {

return next_min_key_ == ustore_key_unknown_k && fetched_offset_ >= fetched_keys_.size();

}

ustore_database_t db_;

ustore_transaction_t txn_;

ustore_collection_t collection_;

ustore_key_t min_key_;

ustore_key_t max_key_;

public:

py_docs_kvrange_t(ustore_database_t db,

ustore_transaction_t txn = nullptr,

ustore_collection_t collection = ustore_collection_main_k,

ustore_key_t min_key = std::numeric_limits<ustore_key_t>::min(),

ustore_key_t max_key = ustore_key_unknown_k) noexcept

: db_(db), txn_(txn), collection_(collection), min_key_(min_key), max_key_(max_key) {}

py_docs_kvrange_t& since(ustore_key_t min_key) noexcept {

min_key_ = min_key;

return *this;

}

py_docs_kvrange_t& until(ustore_key_t max_key) noexcept {

max_key_ = max_key;

return *this;

}

ustore_key_t max_key() noexcept { return max_key_; }

docs_pairs_stream_t begin() noexcept(false) {

docs_pairs_stream_t stream {db_, collection_, docs_pairs_stream_t::default_read_ahead_k, txn_};

status_t status = stream.seek(min_key_);

return stream;

}

std::string json_str;

to_string(val_py, json_str);

ustore_key_t key = py_to_scalar<ustore_key_t>(key_py);

py_collection.native[key].assign(value_view_t(json_str)).throw_unhandled();

if (!PySequence_Check(keys_py) || !PySequence_Check(vals_py))

throw std::invalid_argument("Keys And Vals Must Be Sequence");

std::size_t keys_count = PySequence_Size(keys_py);

if (keys_count != PySequence_Size(vals_py))

throw std::invalid_argument("Keys Count Must Match Values Count");

std::vector<ustore_key_t> keys(keys_count);

std::vector<ustore_length_t> offs(keys_count + 1);

std::string jsons;

offs[0] = 0;

size_t idx = 1;

auto generate_values = [&](py::handle const& obj) {

to_string(obj.ptr(), jsons);

offs[idx] = jsons.size();

++idx;

return dummy_iterator_t {};

};

py_transform_n(keys_py, &py_to_scalar<ustore_key_t>, keys.begin());

py_transform_n(vals_py, generate_values, dummy_iterator_t {});

auto vals_begin = reinterpret_cast<ustore_bytes_ptr_t>(jsons.data());

contents_arg_t values {};

values.offsets_begin = {offs.data(), sizeof(ustore_length_t)};

values.contents_begin = {&vals_begin, 0};

values.count = keys_count;

auto ref = py_collection.native[keys];

ref.assign(values).throw_unhandled();

if (!PySequence_Check(keys_py))

throw std::invalid_argument("Keys Must Be Sequence");

std::vector<ustore_key_t> keys(PySequence_Size(keys_py));

py_transform_n(keys_py, &py_to_scalar<ustore_key_t>, keys.begin());

std::string json_str;

to_string(val_py, json_str);

py_collection.native[keys].assign(value_view_t(json_str)).throw_unhandled();

auto is_single_key = !PySequence_Check(key_py.ptr());

auto func = !is_single_key ? &write_many_docs : &write_one_doc;

return func(py_collection, key_py.ptr(), val_py.ptr());

ustore_key_t key = py_to_scalar<ustore_key_t>(key_py);

auto value = py_collection.native[key].value();

return value->empty() ? py::none {} : py::reinterpret_steal<py::object>(from_json(json_t::parse(*value)));

if (!PySequence_Check(keys_py))

throw std::invalid_argument("Keys Must Be Sequence");

std::vector<ustore_key_t> keys(PySequence_Size(keys_py));

py_transform_n(keys_py, &py_to_scalar<ustore_key_t>, keys.begin());

py::list values(keys.size());

auto maybe_retrieved = py_collection.native[keys].value();

auto const& retrieved = maybe_retrieved.throw_or_ref();

auto it = retrieved.begin();

for (std::size_t i = 0; i != retrieved.size(); ++i)

values[i] = it[i].empty() ? py::none {} : py::reinterpret_steal<py::object>(from_json(json_t::parse(it[i])));

return values;

auto is_single = !PySequence_Check(key_py.ptr());

auto func = is_single ? &read_one_doc : &read_many_docs;

return func(py_collection, key_py.ptr());

auto is_single = !PySequence_Check(key_py.ptr());

auto func = is_single ? &write_one_binary<docs_collection_t> : &write_many_binaries<docs_collection_t>;

return func(py_collection, key_py.ptr(), Py_None);

ustore_key_t key = py_to_scalar<ustore_key_t>(key_py.ptr());

std::string json_str;

to_string(val_py.ptr(), json_str);

py_collection.native[key].merge(value_view_t(json_str));

ustore_key_t key = py_to_scalar<ustore_key_t>(key_py.ptr());

std::string json_str;

to_string(val_py.ptr(), json_str);

py_collection.native[key].patch(value_view_t(json_str));

using py_docs_kvstream_t = py_stream_with_ending_gt<docs_pairs_stream_t>;

auto py_docs_collection = py::class_<py_docs_collection_t>(m, "DocsCollection", py::module_local());

auto py_docs_kvrange = py::class_<py_docs_kvrange_t>(m, "DocsKVRange", py::module_local());

auto py_docs_kvstream = py::class_<py_docs_kvstream_t>(m, "DocsKVStream", py::module_local());

py_docs_collection.def("set", &write_doc);

py_docs_collection.def("get", &read_doc);

py_docs_collection.def("remove", &remove_doc);

py_docs_collection.def("has_key", &has_doc);

py_docs_collection.def("scan", &scan_doc);

py_docs_collection.def("broadcast", &broadcast_doc);

py_docs_collection.def("__setitem__", &write_doc);

py_docs_collection.def("__delitem__", &remove_doc);

py_docs_collection.def("__getitem__", &read_doc);

py_docs_collection.def("__contains__", &has_doc);

py_docs_collection.def("__len__", &get_length<docs_collection_t>);

py_docs_collection.def("clear", [](py_docs_collection_t& collection) { collection.native.clear(); });

py_docs_collection.def("merge", &merge);

py_docs_collection.def("patch", &patch);

py_docs_collection.def_property_readonly("keys", [](py_docs_collection_t& py_collection) {

blobs_range_t members(py_collection.db(), py_collection.txn(), 0, *py_collection.member_collection());

keys_range_t range {members};

return py::cast(std::make_unique<keys_range_t>(range));

});

py_docs_collection.def_property_readonly("items", [](py_docs_collection_t& py_collection) {

py_docs_kvrange_t range(py_collection.db(), py_collection.txn(), *py_collection.member_collection());

return py::cast(std::make_unique<py_docs_kvrange_t>(range));

});

py_docs_kvrange.def("__iter__", [](py_docs_kvrange_t& range) {

docs_pairs_stream_t stream = range.begin();

py_stream_with_ending_gt<docs_pairs_stream_t> wrap {std::move(stream), range.max_key()};

return std::make_unique<py_stream_with_ending_gt<docs_pairs_stream_t>>(std::move(wrap));

});

py_docs_kvrange.def("since", [](py_docs_kvrange_t& range, ustore_key_t key) { return range.since(key); });

py_docs_kvrange.def("until", [](py_docs_kvrange_t& range, ustore_key_t key) { return range.until(key); });

py_docs_kvstream.def("__next__", [](py_docs_kvstream_t& kvstream) {

ustore_key_t key = kvstream.native.key();

if (kvstream.native.is_end() || kvstream.stop)

throw py::stop_iteration();

kvstream.stop = kvstream.terminal == key;

value_view_t value_view = kvstream.native.value();

++kvstream.native;

return py::make_tuple(key, py::reinterpret_steal<py::object>(from_json(json_t::parse(value_view))));

});

}