typed_txn_btree.h

#ifndef _NDB_TYPED_TXN_BTREE_H_
#define _NDB_TYPED_TXN_BTREE_H_

#include "base_txn_btree.h"
#include "txn_btree.h"
#include "record/cursor.h"

template <typename Schema>
struct typed_txn_btree_ {

  typedef typename Schema::base_type base_type;
  typedef typename Schema::key_type key_type;
  typedef typename Schema::value_type value_type;
  typedef typename Schema::value_descriptor_type value_descriptor_type;
  typedef typename Schema::key_encoder_type key_encoder_type;
  typedef typename Schema::value_encoder_type value_encoder_type;

  static_assert(value_descriptor_type::nfields() <= 64, "xx");
  static const uint64_t AllFieldsMask = (1UL << value_descriptor_type::nfields()) - 1;

  static inline constexpr bool
  IsAllFields(uint64_t m)
  {
    return (m & AllFieldsMask) == AllFieldsMask;
  }

  class key_reader {
  public:
    constexpr key_reader(bool no_key_results) : no_key_results(no_key_results) {}
    inline const key_type &
    operator()(const std::string &s)
    {
      const typename Schema::key_encoder_type key_encoder;
      if (!no_key_results)
        key_encoder.read(s, &k);
      return k;
    }
    inline const key_type &
    operator()(lcdf::Str s)
    {
      const typename Schema::key_encoder_type key_encoder;
      if (!no_key_results)
        key_encoder.read(s, &k);
      return k;
    }
  private:
    key_type k;
    bool no_key_results;
  };

  static inline bool
  do_record_read(const uint8_t *data, size_t sz, uint64_t fields_mask, value_type *v)
  {
    if (IsAllFields(fields_mask)) {
      // read the entire record
      const value_encoder_type value_encoder;
      return value_encoder.failsafe_read(data, sz, v);
    } else {
      // pick individual fields
      read_record_cursor<base_type> r(data, sz);
      for (uint64_t i = 0; i < value_descriptor_type::nfields(); i++) {
        if ((1UL << i) & fields_mask) {
          r.skip_to(i);
          if (unlikely(!r.read_current_and_advance(v)))
            return false;
        }
      }
      return true;
    }
  }

  class single_value_reader {
  public:
    typedef typename Schema::value_type value_type;

    constexpr single_value_reader(value_type &v, uint64_t fields_mask)
      : v(&v), fields_mask(fields_mask) {}

    template <typename StringAllocator>
    inline bool
    operator()(const uint8_t *data, size_t sz, StringAllocator &sa)
    {
      return do_record_read(data, sz, fields_mask, v);
    }

    inline value_type &
    results()
    {
      return *v;
    }

    inline const value_type &
    results() const
    {
      return *v;
    }

    template <typename StringAllocator>
    inline void
    dup(const value_type &vdup, StringAllocator &sa)
    {
      *v = vdup;
    }

  private:
    value_type *v;
    uint64_t fields_mask;
  };

  class value_reader {
  public:
    typedef typename Schema::value_type value_type;

    constexpr value_reader(uint64_t fields_mask) : fields_mask(fields_mask) {}

    template <typename StringAllocator>
    inline bool
    operator()(const uint8_t *data, size_t sz, StringAllocator &sa)
    {
      return do_record_read(data, sz, fields_mask, &v);
    }

    inline value_type &
    results()
    {
      return v;
    }

    inline const value_type &
    results() const
    {
      return v;
    }

    template <typename StringAllocator>
    inline void
    dup(const value_type &vdup, StringAllocator &sa)
    {
      v = vdup;
    }

  private:
    value_type v;
    uint64_t fields_mask;
  };

  class key_writer {
  public:
    constexpr key_writer(const key_type *k) : k(k) {}

    template <typename StringAllocator>
    inline const std::string *
    fully_materialize(bool stable_input, StringAllocator &sa)
    {
      if (!k)
        return nullptr;
      std::string * const ret = sa();
      const key_encoder_type key_encoder;
      key_encoder.write(*ret, k);
      return ret;
    }
  private:
    const key_type *k;
  };

  static inline size_t
  compute_needed_standalone(
    const value_type *v, uint64_t fields,
    const uint8_t *buf, size_t sz)
  {
    if (fields == 0) {
      // delete
      INVARIANT(!v);
      return 0;
    }
    INVARIANT(v);
    if (sz == 0) {
      // new record (insert)
      INVARIANT(IsAllFields(fields));
      const value_encoder_type value_encoder;
      return value_encoder.nbytes(v);
    }

    ssize_t new_updates_sum = 0;
    for (uint64_t i = 0; i < value_descriptor_type::nfields(); i++) {
      if ((1UL << i) & fields) {
        const uint8_t * px = reinterpret_cast<const uint8_t *>(v) +
          value_descriptor_type::cstruct_offsetof(i);
        new_updates_sum += value_descriptor_type::nbytes_fn(i)(px);
      }
    }

    // XXX: should try to cache pointers discovered by read_record_cursor
    ssize_t old_updates_sum = 0;
    read_record_cursor<base_type> rc(buf, sz);
    for (uint64_t i = 0; i < value_descriptor_type::nfields(); i++) {
      if ((1UL << i) & fields) {
        rc.skip_to(i);
        const size_t sz = rc.read_current_raw_size_and_advance();
        INVARIANT(sz);
        old_updates_sum += sz;
      }
    }

    // XXX: see if approximate version works almost as well (approx version is
    // to assume that each field has the minimum possible size, which is
    // overly conservative but correct)

    const ssize_t ret = static_cast<ssize_t>(sz) - old_updates_sum + new_updates_sum;
    INVARIANT(ret > 0);
    return ret;
  }

  // how many bytes do we need to encode a delta record
  static inline size_t
  compute_needed_delta_standalone(
      const value_type *v, uint64_t fields)
  {
    size_t size_needed = 0;
    size_needed += sizeof(uint64_t);
    if (fields == 0) {
      // delete
      INVARIANT(!v);
      return size_needed;
    }
    INVARIANT(v);
    if (IsAllFields(fields)) {
      // new record (insert)
      const value_encoder_type value_encoder;
      size_needed += value_encoder.nbytes(v);
      return size_needed;
    }

    for (uint64_t i = 0; i < value_descriptor_type::nfields(); i++) {
      if ((1UL << i) & fields) {
        const uint8_t * px = reinterpret_cast<const uint8_t *>(v) +
          value_descriptor_type::cstruct_offsetof(i);
        size_needed += value_descriptor_type::nbytes_fn(i)(px);
      }
    }

    return size_needed;
  }

  static inline void
  do_write_standalone(
      const value_type *v, uint64_t fields,
      uint8_t *buf, size_t sz)
  {
    if (fields == 0) {
      // no-op for delete
      INVARIANT(!v);
      return;
    }
    if (IsAllFields(fields)) {
      // special case, just use the standard encoder (faster)
      // because it's straight-line w/ no branching
      const value_encoder_type value_encoder;
      value_encoder.write(buf, v);
      return;
    }
    write_record_cursor<base_type> wc(buf);
    for (uint64_t i = 0; i < value_descriptor_type::nfields(); i++) {
      if ((1UL << i) & fields) {
        wc.skip_to(i);
        wc.write_current_and_advance(v, nullptr);
      }
    }
  }

  static inline void
  do_delta_write_standalone(
      const value_type *v, uint64_t fields,
      uint8_t *buf, size_t sz)
  {
    serializer<uint64_t, false> s_uint64_t;

#ifdef CHECK_INVARIANTS
    const uint8_t * const orig_buf = buf;
#endif

    buf = s_uint64_t.write(buf, fields);
    if (fields == 0) {
      // no-op for delete
      INVARIANT(!v);
      return;
    }
    if (IsAllFields(fields)) {
      // special case, just use the standard encoder (faster)
      // because it's straight-line w/ no branching
      const value_encoder_type value_encoder;
      value_encoder.write(buf, v);
      return;
    }
    for (uint64_t i = 0; i < value_descriptor_type::nfields(); i++) {
      if ((1UL << i) & fields) {
        const uint8_t * px = reinterpret_cast<const uint8_t *>(v) +
          value_descriptor_type::cstruct_offsetof(i);
        buf = value_descriptor_type::write_fn(i)(buf, px);
      }
    }

    INVARIANT(buf - orig_buf == ptrdiff_t(sz));
  }

  template <uint64_t Fields>
  static inline size_t
  tuple_writer(dbtuple::TupleWriterMode mode, const void *v, uint8_t *p, size_t sz)
  {
    const value_type *vx = reinterpret_cast<const value_type *>(v);
    switch (mode) {
    case dbtuple::TUPLE_WRITER_NEEDS_OLD_VALUE:
      return 1;
    case dbtuple::TUPLE_WRITER_COMPUTE_NEEDED:
      return compute_needed_standalone(vx, Fields, p, sz);
    case dbtuple::TUPLE_WRITER_COMPUTE_DELTA_NEEDED:
      return compute_needed_delta_standalone(vx, Fields);
    case dbtuple::TUPLE_WRITER_DO_WRITE:
      do_write_standalone(vx, Fields, p, sz);
      return 0;
    case dbtuple::TUPLE_WRITER_DO_DELTA_WRITE:
      do_delta_write_standalone(vx, Fields, p, sz);
      return 0;
    }
    ALWAYS_ASSERT(false);
    return 0;
  }

  class value_writer {
  public:
    constexpr value_writer(const value_type *v, uint64_t fields)
      : v(v), fields(fields) {}

    // old version of record is stored at
    // [buf, buf+sz).
    //
    // compute the new required size for the update
    inline size_t
    compute_needed(const uint8_t *buf, size_t sz)
    {
      return compute_needed_standalone(v, fields, buf, sz);
    }

    template <typename StringAllocator>
    inline const std::string *
    fully_materialize(bool stable_input, StringAllocator &sa)
    {
      INVARIANT(IsAllFields(fields) || fields == 0);
      if (fields == 0) {
        // delete
        INVARIANT(!v);
        return nullptr;
      }
      std::string * const ret = sa();
      const value_encoder_type value_encoder;
      value_encoder.write(*ret, v);
      return ret;
    }

    // the old value lives in [buf, buf+sz), but [buf, buf+compute_needed())
    // is valid memory to write to
    inline void
    operator()(uint8_t *buf, size_t sz)
    {
      do_write_standalone(v, fields, buf, sz);
    }

  private:
    const value_type *v;
    uint64_t fields;
  };

  typedef key_type Key;
  typedef key_writer KeyWriter;
  typedef value_type Value;
  typedef value_writer ValueWriter;
  typedef uint64_t ValueInfo;

  //typedef key_reader KeyReader;
  //typedef single_value_reader SingleValueReader;
  //typedef value_reader ValueReader;

};

template <template <typename> class Transaction, typename Schema>
class typed_txn_btree : public base_txn_btree<Transaction, typed_txn_btree_<Schema>> {
  typedef base_txn_btree<Transaction, typed_txn_btree_<Schema>> super_type;
public:

  typedef typename super_type::string_type string_type;
  typedef typename super_type::size_type size_type;

  typedef typename Schema::base_type base_type;
  typedef typename Schema::key_type key_type;
  typedef typename Schema::value_type value_type;
  typedef typename Schema::value_descriptor_type value_descriptor_type;
  typedef typename Schema::key_encoder_type key_encoder_type;
  typedef typename Schema::value_encoder_type value_encoder_type;

private:

  typedef txn_btree_::key_reader bytes_key_reader;
  typedef txn_btree_::single_value_reader bytes_single_value_reader;
  typedef txn_btree_::value_reader bytes_value_reader;

  typedef
    typename typed_txn_btree_<Schema>::key_writer
    key_writer;

  typedef
    typename typed_txn_btree_<Schema>::key_reader
    key_reader;
  typedef
    typename typed_txn_btree_<Schema>::single_value_reader
    single_value_reader;
  typedef
    typename typed_txn_btree_<Schema>::value_reader
    value_reader;

  template <typename Traits>
  static constexpr inline bool
  IsSupportable()
  {
    return Traits::stable_input_memory ||
      (mass::is_trivially_copyable<key_type>::value &&
       mass::is_trivially_destructible<key_type>::value &&
       mass::is_trivially_copyable<value_type>::value &&
       mass::is_trivially_destructible<value_type>::value);
  }

public:

  static const uint64_t AllFieldsMask = typed_txn_btree_<Schema>::AllFieldsMask;
  typedef util::Fields<AllFieldsMask> AllFields;

  struct search_range_callback {
  public:
    virtual ~search_range_callback() {}
    virtual bool invoke(const key_type &k, const value_type &v) = 0;
  };

  struct bytes_search_range_callback {
  public:
    virtual ~bytes_search_range_callback() {}
    virtual bool invoke(const string_type &k, const string_type &v) = 0;
  };

  typed_txn_btree(size_type value_size_hint = 128,
                  bool mostly_append = false,
                  const std::string &name = "<unknown>")
    : super_type(value_size_hint, mostly_append, name)
  {}

  template <typename Traits, typename FieldsMask = AllFields>
  inline bool search(
      Transaction<Traits> &t, const key_type &k, value_type &v,
      FieldsMask fm = FieldsMask());

  template <typename Traits, typename FieldsMask = AllFields>
  inline void search_range_call(
      Transaction<Traits> &t, const key_type &lower, const key_type *upper,
      search_range_callback &callback,
      bool no_key_results = false /* skip decoding of keys? */,
      FieldsMask fm = FieldsMask());

  // a lower-level variant which does not bother to decode the key/values
  template <typename Traits>
  inline void bytes_search_range_call(
      Transaction<Traits> &t, const key_type &lower, const key_type *upper,
      bytes_search_range_callback &callback,
      size_type value_fields_prefix = std::numeric_limits<size_type>::max());

  template <typename Traits, typename FieldsMask = AllFields>
  inline void put(
      Transaction<Traits> &t, const key_type &k, const value_type &v,
      FieldsMask fm = FieldsMask());

  template <typename Traits>
  inline void insert(
      Transaction<Traits> &t, const key_type &k, const value_type &v);

  template <typename Traits>
  inline void remove(
      Transaction<Traits> &t, const key_type &k);

private:

  template <typename Traits>
  static inline const std::string *
  stablize(Transaction<Traits> &t, const key_type &k)
  {
    key_writer writer(&k);
    return writer.fully_materialize(
        Traits::stable_input_memory, t.string_allocator());
  }

  template <typename Traits>
  static inline const value_type *
  stablize(Transaction<Traits> &t, const value_type &v)
  {
    if (Traits::stable_input_memory)
      return &v;
    std::string * const px = t.string_allocator()();
    px->assign(reinterpret_cast<const char *>(&v), sizeof(v));
    return reinterpret_cast<const value_type *>(px->data());
  }

  key_encoder_type key_encoder;
  value_encoder_type value_encoder;
};

template <template <typename> class Transaction, typename Schema>
template <typename Traits, typename FieldsMask>
bool
typed_txn_btree<Transaction, Schema>::search(
    Transaction<Traits> &t, const key_type &k, value_type &v,
    FieldsMask fm)
{
  // XXX: template single_value_reader with mask
  single_value_reader vr(v, FieldsMask::value);
  return this->do_search(t, k, vr);
}

template <template <typename> class Transaction, typename Schema>
template <typename Traits, typename FieldsMask>
void
typed_txn_btree<Transaction, Schema>::search_range_call(
    Transaction<Traits> &t,
    const key_type &lower, const key_type *upper,
    search_range_callback &callback,
    bool no_key_results,
    FieldsMask fm)
{
  key_reader kr(no_key_results);
  value_reader vr(FieldsMask::value);
  this->do_search_range_call(t, lower, upper, callback, kr, vr);
}

template <template <typename> class Transaction, typename Schema>
template <typename Traits>
void
typed_txn_btree<Transaction, Schema>::bytes_search_range_call(
    Transaction<Traits> &t, const key_type &lower, const key_type *upper,
    bytes_search_range_callback &callback,
    size_type value_fields_prefix)
{
  const value_encoder_type value_encoder;
  const size_t max_bytes_read =
    value_encoder.encode_max_nbytes_prefix(value_fields_prefix);
  bytes_key_reader kr;
  bytes_value_reader vr(max_bytes_read);
  this->do_search_range_call(t, lower, upper, callback, kr, vr);
}

template <template <typename> class Transaction, typename Schema>
template <typename Traits, typename FieldsMask>
void
typed_txn_btree<Transaction, Schema>::put(
    Transaction<Traits> &t, const key_type &k, const value_type &v, FieldsMask fm)
{
  static_assert(IsSupportable<Traits>(), "xx");
  const dbtuple::tuple_writer_t tw =
    &typed_txn_btree_<Schema>::template tuple_writer<FieldsMask::value>;
  this->do_tree_put(t, stablize(t, k), stablize(t, v), tw, false);
}

template <template <typename> class Transaction, typename Schema>
template <typename Traits>
void
typed_txn_btree<Transaction, Schema>::insert(
    Transaction<Traits> &t, const key_type &k, const value_type &v)
{
  static_assert(IsSupportable<Traits>(), "xx");
  const dbtuple::tuple_writer_t tw =
    &typed_txn_btree_<Schema>::template tuple_writer<AllFieldsMask>;
  this->do_tree_put(t, stablize(t, k), stablize(t, v), tw, true);
}

template <template <typename> class Transaction, typename Schema>
template <typename Traits>
void
typed_txn_btree<Transaction, Schema>::remove(
    Transaction<Traits> &t, const key_type &k)
{
  static_assert(IsSupportable<Traits>(), "xx");
  const dbtuple::tuple_writer_t tw =
    &typed_txn_btree_<Schema>::template tuple_writer<0>;
  this->do_tree_put(t, stablize(t, k), nullptr, tw, false);
}

#endif /* _NDB_TYPED_TXN_BTREE_H_ */