compound: Introduce legacy format adaptors

compound_compat.hh

@@ -0,0 +1,175 @@
+/*
+ * Copyright 2015 Cloudius Systems
+ */
+
+#pragma once
+
+#include "compound.hh"
+
+//
+// This header provides adaptors between the representation used by our compound_type<>
+// and representation used by Origin.
+//
+// For single-component keys the legacy representation is equivalent
+// to the only component's serialized form. For composite keys it the following
+// (See org.apache.cassandra.db.marshal.CompositeType):
+//
+//   <representation> ::= ( <component> )+
+//   <component>      ::= <length> <value> <EOC>
+//   <length>         ::= <uint16_t>
+//   <EOC>            ::= <uint8_t>
+//
+//  <value> is component's value in serialized form. <EOC> is always 0 for partition key.
+//
+
+// Given a representation serialized using @CompoundType, provides a view on the
+// representation of the same components as they would be serialized by Origin.
+//
+// The view is exposed in a form of a byte range. For example of use see to_legacy() function.
+template <typename CompoundType>
+class legacy_compound_view {
+    static_assert(!CompoundType::is_prefixable, "Legacy view not defined for prefixes");
+    CompoundType& _type;
+    bytes_view _packed;
+public:
+    legacy_compound_view(CompoundType& c, bytes_view packed)
+        : _type(c)
+        , _packed(packed)
+    { }
+
+    class iterator : public std::iterator<std::input_iterator_tag, bytes::value_type> {
+        bool _singular;
+        // Offset within virtual output space of a component.
+        //
+        // Offset: -2             -1             0  ...  LEN-1 LEN
+        // Field:  [ length MSB ] [ length LSB ] [   VALUE   ] [ EOC ]
+        //
+        int32_t _offset;
+        typename CompoundType::iterator _i;
+    public:
+        struct end_tag {};
+
+        iterator(const legacy_compound_view& v)
+            : _singular(v._type.is_singular())
+            , _offset(_singular ? 0 : -2)
+            , _i(v._type.begin(v._packed))
+        { }
+
+        iterator(const legacy_compound_view& v, end_tag)
+            : _offset(-2)
+            , _i(v._type.end(v._packed))
+        { }
+
+        value_type operator*() const {
+            int32_t component_size = _i->size();
+            if (_offset == -2) {
+                return (component_size >> 8) & 0xff;
+            } else if (_offset == -1) {
+                return component_size & 0xff;
+            } else if (_offset < component_size) {
+                return (*_i)[_offset];
+            } else { // _offset == component_size
+                return 0; // EOC field
+            }
+        }
+
+        iterator& operator++() {
+            auto component_size = (int32_t) _i->size();
+            if (_offset < component_size
+                // When _singular, we skip the EOC byte.
+                && (!_singular || _offset != (component_size - 1)))
+            {
+                ++_offset;
+            } else {
+                ++_i;
+                _offset = -2;
+            }
+            return *this;
+        }
+
+        bool operator==(const iterator& other) const {
+            return _offset == other._offset && other._i == _i;
+        }
+
+        bool operator!=(const iterator& other) const {
+            return !(*this == other);
+        }
+    };
+
+    // A trichotomic comparator defined on @CompoundType representations which
+    // orders them according to lexicographical ordering of their corresponding
+    // legacy representations.
+    //
+    //   tri_comparator(t)(k1, k2)
+    //
+    // ...is equivalent to:
+    //
+    //   compare_unsigned(to_legacy(t, k1), to_legacy(t, k2))
+    //
+    // ...but more efficient.
+    //
+    struct tri_comparator {
+        const CompoundType& _type;
+
+        tri_comparator(const CompoundType& type)
+            : _type(type)
+        { }
+
+        tri_comparator(tri_comparator&& other)
+            : _type(other._type)
+        { }
+
+        tri_comparator& operator=(tri_comparator&& other) {
+            this->~tri_comparator();
+            new (this) tri_comparator(std::move(other));
+            return *this;
+        }
+
+        // @k1 and @k2 must be serialized using @type, which was passed to the constructor.
+        int operator()(bytes_view k1, bytes_view k2) const {
+            if (_type.is_singular()) {
+                return compare_unsigned(*_type.begin(k1), *_type.begin(k2));
+            }
+            return lexicographical_tri_compare(
+                _type.begin(k1), _type.end(k1),
+                _type.begin(k2), _type.end(k2),
+                [] (const bytes_view& c1, const bytes_view& c2) -> int {
+                    if (c1.size() != c2.size()) {
+                        return c1.size() < c2.size() ? -1 : 1;
+                    }
+                    return memcmp(c1.begin(), c2.begin(), c1.size());
+                });
+        }
+    };
+
+    // Equivalent to std::distance(begin(), end()), but computes faster
+    size_t size() const {
+        if (_type.is_singular()) {
+            return _type.begin(_packed)->size();
+        }
+        size_t s = 0;
+        for (auto&& component : _type.components(_packed)) {
+            s += 2 /* length field */ + component.size() + 1 /* EOC */;
+        }
+        return s;
+    }
+
+    iterator begin() const {
+        return iterator(*this);
+    }
+
+    iterator end() const {
+        return iterator(*this, typename iterator::end_tag());
+    }
+};
+
+// Converts compound_type<> representation to legacy representation
+// @packed is assumed to be serialized using supplied @type.
+template <typename CompoundType>
+static inline
+bytes to_legacy(CompoundType& type, bytes_view packed) {
+    legacy_compound_view<CompoundType> lv(type, packed);
+    bytes legacy_form(bytes::initialized_later(), lv.size());
+    std::copy(lv.begin(), lv.end(), legacy_form.begin());
+    return legacy_form;
+}