Loading...
Searching...
No Matches
stream_query.hxx
/* Definition of the pqxx::internal::stream_query class.
*
* Enables optimized batch reads from a database table.
*
* DO NOT INCLUDE THIS FILE DIRECTLY; include pqxx/stream_query instead.
*
* Copyright (c) 2000-2024, Jeroen T. Vermeulen.
*
* See COPYING for copyright license. If you did not receive a file called
* COPYING with this source code, please notify the distributor of this
* mistake, or contact the author.
*/
#ifndef PQXX_H_STREAM_QUERY
#define PQXX_H_STREAM_QUERY
#if !defined(PQXX_HEADER_PRE)
# error "Include libpqxx headers as <pqxx/header>, not <pqxx/header.hxx>."
#endif
#include <cassert>
#include <functional>
#include <variant>
#include "pqxx/connection.hxx"
#include "pqxx/except.hxx"
#include "pqxx/internal/concat.hxx"
#include "pqxx/internal/encoding_group.hxx"
#include "pqxx/internal/encodings.hxx"
#include "pqxx/internal/gates/connection-stream_from.hxx"
#include "pqxx/internal/stream_iterator.hxx"
#include "pqxx/separated_list.hxx"
#include "pqxx/transaction_base.hxx"
#include "pqxx/transaction_focus.hxx"
#include "pqxx/util.hxx"
namespace pqxx
{
class transaction_base;
} // namespace pqxx
namespace pqxx::internal
{
class stream_query_end_iterator
{};
// C++20: Can we use generators, and maybe get speedup from HALO?
template<typename... TYPE> class stream_query : transaction_focus
{
public:
using line_handle = std::unique_ptr<char, void (*)(void const *)>;
inline stream_query(transaction_base &tx, std::string_view query);
inline stream_query(
transaction_base &tx, std::string_view query, params const &);
stream_query(stream_query &&) = delete;
stream_query &operator=(stream_query &&) = delete;
~stream_query() noexcept
{
try
{
close();
}
catch (std::exception const &e)
{
reg_pending_error(e.what());
}
}
bool done() const & noexcept { return m_char_finder == nullptr; }
inline auto begin() &;
auto end() const & { return stream_query_end_iterator{}; }
std::tuple<TYPE...> parse_line(zview line) &
{
assert(not done());
auto const line_size{std::size(line)};
// This function uses m_row as a buffer, across calls. The only reason for
// it to carry over across calls is to avoid reallocation.
// Make room for unescaping the line. It's a pessimistic size.
// Unusually, we're storing terminating zeroes *inside* the string.
// This is the only place where we modify m_row. MAKE SURE THE BUFFER DOES
// NOT GET RESIZED while we're working, because we're working with views
// into its buffer.
m_row.resize(line_size + 1);
std::size_t offset{0u};
char *write{m_row.data()};
// DO NOT shrink m_row to fit. We're carrying views pointing into the
// buffer. (Also, how useful would shrinking really be?)
// Folding expression: scan and unescape each field, and convert it to its
// requested type.
std::tuple<TYPE...> data{parse_field<TYPE>(line, offset, write)...};
assert(offset == line_size + 1u);
return data;
}
std::pair<line_handle, std::size_t> read_line() &;
private:
static inline char_finder_func *get_finder(transaction_base const &tx);
std::tuple<std::size_t, char *, zview>
read_field(zview line, std::size_t offset, char *write)
{
#if !defined(NDEBUG)
auto const line_size{std::size(line)};
#endif
assert(offset <= line_size);
char const *lp{std::data(line)};
// The COPY line now ends in a tab. (We replace the trailing newline with
// that to simplify the loop here.)
assert(lp[line_size] == '\t');
assert(lp[line_size + 1] == '\0');
if ((lp[offset] == '\\') and (lp[offset + 1] == 'N'))
{
// Null field. Consume the "\N" and the field separator.
offset += 3;
assert(offset <= (line_size + 1));
assert(lp[offset - 1] == '\t');
// Return a null value. There's nothing to write into m_row.
return {offset, write, {}};
}
// Beginning of the field text in the row buffer.
char const *const field_begin{write};
// We're relying on several assumptions just for making the main loop
// condition work:
// * The COPY line ends in a newline.
// * Multibyte characters never start with an ASCII-range byte.
// * We can index a view beyond its bounds (but within its address space).
//
// Effectively, the newline acts as a final field separator.
while (lp[offset] != '\t')
{
assert(lp[offset] != '\0');
// Beginning of the next character of interest (or the end of the line).
auto const stop_char{m_char_finder(line, offset)};
PQXX_ASSUME(stop_char > offset);
assert(stop_char < (line_size + 1));
// Copy the text we have so far. It's got no special characters in it.
std::memcpy(write, &lp[offset], stop_char - offset);
write += (stop_char - offset);
offset = stop_char;
// We're still within the line.
char const special{lp[offset]};
if (special == '\\')
{
// Escape sequence.
// Consume the backslash.
++offset;
assert(offset < line_size);
// The database will only escape ASCII characters, so we assume that
// we're dealing with a single-byte character.
char const escaped{lp[offset]};
assert((escaped >> 7) == 0);
++offset;
*write++ = unescape_char(escaped);
}
else
{
// Field separator. Fall out of the loop.
assert(special == '\t');
}
}
// Hit the end of the field.
assert(lp[offset] == '\t');
*write = '\0';
++write;
++offset;
return {offset, write, {field_begin, write - field_begin - 1}};
}
template<typename TARGET>
TARGET parse_field(zview line, std::size_t &offset, char *&write)
{
using field_type = strip_t<TARGET>;
using nullity = nullness<field_type>;
assert(offset <= std::size(line));
auto [new_offset, new_write, text]{read_field(line, offset, write)};
PQXX_ASSUME(new_offset > offset);
PQXX_ASSUME(new_write >= write);
offset = new_offset;
write = new_write;
if constexpr (nullity::always_null)
{
if (std::data(text) != nullptr)
throw conversion_error{concat(
"Streaming a non-null value into a ", type_name<field_type>,
", which must always be null.")};
}
else if (std::data(text) == nullptr)
{
if constexpr (nullity::has_null)
return nullity::null();
else
internal::throw_null_conversion(type_name<field_type>);
}
else
{
// Don't ever try to convert a non-null value to nullptr_t!
return from_string<field_type>(text);
}
}
void close() noexcept
{
if (not done())
{
m_char_finder = nullptr;
unregister_me();
}
}
char_finder_func *m_char_finder;
std::string m_row;
};
} // namespace pqxx::internal
#endif
pqxx::internal::stream_query
Stream query results from the database. Used by transaction_base::stream.
Definition stream_query.hxx:80
pqxx::internal::stream_query::done
bool done() const &noexcept
Has this stream reached the end of its data?
Definition stream_query.hxx:106
pqxx::internal::stream_query::parse_line
std::tuple< TYPE... > parse_line(zview line) &
Parse and convert the latest line of data we received.
Definition stream_query.hxx:118
pqxx::internal::stream_query::begin
auto begin() &
Begin iterator. Only for use by "range for.".
Definition stream_query_impl.hxx:166
pqxx::internal::stream_query::read_line
std::pair< line_handle, std::size_t > read_line() &
Read a COPY line from the server.
Definition stream_query_impl.hxx:174
pqxx::internal::stream_query::end
auto end() const &
End iterator. Only for use by "range for.".
Definition stream_query.hxx:115
pqxx::params
Build a parameter list for a parameterised or prepared statement.
Definition params.hxx:33
pqxx::transaction_focus
Base class for things that monopolise a transaction's attention.
Definition transaction_focus.hxx:29
pqxx::zview
Marker-type wrapper: zero-terminated std::string_view.
Definition zview.hxx:38
pqxx::transaction_base
Interface definition (and common code) for "transaction" classes.
Definition transaction_base.hxx:151
pqxx::internal
Internal items for libpqxx' own use. Do not use these yourself.
Definition encodings.cxx:33
pqxx::internal::concat
std::string concat(TYPE... item)
Efficiently combine a bunch of items into one big string.
Definition concat.hxx:31
pqxx::internal::throw_null_conversion
void throw_null_conversion(std::string const &type)
Throw exception for attempt to convert SQL NULL to given type.
Definition strconv.cxx:248
pqxx::internal::char_finder_func
std::size_t(std::string_view haystack, std::size_t start) char_finder_func
Function type: "find first occurrence of specific any of ASCII characters.".
Definition encoding_group.hxx:71
pqxx::internal::unescape_char
constexpr char unescape_char(char escaped) noexcept
Return original byte for escaped character.
Definition util.hxx:633
pqxx::internal::stream_query_end_iterator
The end() iterator for a stream_query.
Definition stream_query.hxx:47
pqxx
The home of all libpqxx classes, functions, templates, etc.
Definition array.cxx:27