stream_query.hxx

Go to the documentation of this file.

/* Definition of the pqxx::internal::stream_query class.
 *
 * Enables optimized batch reads from a database table.
 *
 * DO NOT INCLUDE THIS FILE DIRECTLY; include pqxx/transaction_base instead.
 *
 * Copyright (c) 2000-2026, 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_INTERNAL_STREAM_QUERY_HXX
#define PQXX_INTERNAL_STREAM_QUERY_HXX
 
#if !defined(PQXX_HEADER_PRE)
#  error "Include libpqxx headers as <pqxx/header>, not <pqxx/header.hxx>."
#endif
 
#include "pqxx/internal/encodings.hxx"
#include "pqxx/internal/gates/connection-stream_from.hxx"
#include "pqxx/transaction_focus.hxx"
 
 
namespace pqxx
{
class transaction_base;
} // namespace pqxx
 
 
namespace pqxx::internal
{
class stream_query_end_iterator
{};
 
 
// TODO: Can we use generators, and maybe get speedup from HALO?
 
template<typename... TYPE> class stream_query final : transaction_focus
{
public:
  using line_handle = std::unique_ptr<char[], void (*)(void const *)>;
 
  inline stream_query(
    transaction_base &tx, std::string_view query, conversion_context c);
 
  stream_query(stream_query const &) = delete;
  stream_query(stream_query &&) = delete;
  stream_query &operator=(stream_query const &) = delete;
  stream_query &operator=(stream_query &&) = delete;
 
  ~stream_query() noexcept
  {
    try
    {
      close();
    }
    catch (std::exception const &e)
    {
      reg_pending_error(e.what(), sl::current());
    }
  }
 
  [[nodiscard]] bool done() const & noexcept
  {
    return m_char_finder == nullptr;
  }
 
  [[nodiscard]] inline auto begin() &;
 
 
  [[nodiscard]] auto end() const & { return stream_query_end_iterator{}; }
 
  std::tuple<TYPE...> parse_line(std::string_view 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, m_ctx)...};
 
    assert(offset == line_size + 1u);
    return data;
  }
 
  std::pair<line_handle, std::size_t> read_line(sl) &;
 
private:
 
  PQXX_PURE PQXX_RETURNS_NONNULL static inline char_finder_func *
  get_finder(transaction_base const &tx, sl);
 
 
  std::tuple<std::size_t, char *, std::string_view>
  read_field(std::string_view line, std::size_t offset, char *write, ctx c)
  {
#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).
      // It may be right where we start searching, and this won't loop forever
      // since the previous iteration (if any) put us right _after_ the
      // previous character of interest.
      auto const stop_char{m_char_finder(line, offset, c.loc)};
      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]};
 
        // I think this is a valid way to check for the high bit: the shift
        // may be signed or unsigned (implementation-defined for char), but
        // either way we get a zero if the bit is clear or nonzero if it's set.
        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, static_cast<std::size_t>(write - field_begin - 1)}};
  }
 
 
  template<typename TARGET>
  TARGET
  parse_field(std::string_view line, std::size_t &offset, char *&write, ctx c)
  {
    using field_type = std::remove_cvref_t<TARGET>;
 
    assert(offset <= std::size(line));
 
    auto [new_offset, new_write, text]{read_field(line, offset, write, c)};
    PQXX_ASSUME(new_offset > offset);
    PQXX_ASSUME(new_write >= write);
    offset = new_offset;
    write = new_write;
    if constexpr (pqxx::always_null<TARGET>())
    {
      if (std::data(text) != nullptr)
        throw conversion_error{std::format(
          "Streaming a non-null value into a {}, which must always be null.",
          name_type<field_type>())};
    }
    else if (std::data(text) == nullptr)
    {
      if constexpr (has_null<TARGET>())
        return make_null<TARGET>();
      else
        internal::throw_null_conversion(name_type<field_type>(), c.loc);
    }
    else [[likely]]
    {
      // Don't ever try to convert a non-null value to nullptr_t!
      return from_string<field_type>(text, c);
    }
  }
 
  void close() noexcept
  {
    if (not done())
    {
      m_char_finder = nullptr;
      unregister_me();
    }
  }
 
 
  char_finder_func *m_char_finder;
 
 
  std::string m_row;
 
  conversion_context const m_ctx;
};
} // namespace pqxx::internal
#endif