Merge "Support clustering order" from Glauber

"This is the work required to support the clustering order statement.

There is no work needed from the sstables side, because it should just
write whatever we are given by the memtable layer, in the order established
by the memtable layer.

So the work is basically done in the types-level, to support "reversed types",
one concept from Origin we were missing"

compound.hh

@@ -40,6 +40,7 @@ private:
     const std::vector<data_type> _types;
     const bool _byte_order_equal;
     const bool _byte_order_comparable;
+    const bool _is_reversed;
 public:
     static constexpr bool is_prefixable = AllowPrefixes == allow_prefixes::yes;
     using prefix_type = compound_type<allow_prefixes::yes>;
@@ -51,6 +52,7 @@ public:
                 return t->is_byte_order_equal();
             }))
         , _byte_order_comparable(_types.size() == 1 && _types[0]->is_byte_order_comparable())
+        , _is_reversed(_types.size() == 1 && _types[0]->is_reversed())
     { }
 
     compound_type(compound_type&&) = default;
@@ -246,7 +248,11 @@ public:
     }
     int compare(bytes_view b1, bytes_view b2) {
         if (_byte_order_comparable) {
-            return compare_unsigned(b1, b2);
+            if (_is_reversed) {
+                return compare_unsigned(b2, b1);
+            } else {
+                return compare_unsigned(b1, b2);
+            }
         }
         return lexicographical_tri_compare(_types.begin(), _types.end(),
             begin(b1), end(b1), begin(b2), end(b2), [] (auto&& type, auto&& v1, auto&& v2) {

cql3/statements/create_table_statement.cc

@@ -305,29 +305,29 @@ create_table_statement::raw_statement::raw_statement(::shared_ptr<cf_name> name,
             ? CounterColumnType.instance
             : BytesType.instance;
     }
-
+#endif
 
     // If we give a clustering order, we must explicitly do so for all aliases and in the order of the PK
-    if (!definedOrdering.isEmpty())
-    {
-        if (definedOrdering.size() > columnAliases.size())
-            throw new InvalidRequestException("Only clustering key columns can be defined in CLUSTERING ORDER directive");
+    if (!_defined_ordering.empty()) {
+        if (_defined_ordering.size() > _column_aliases.size()) {
+            throw exceptions::invalid_request_exception("Only clustering key columns can be defined in CLUSTERING ORDER directive");
+        }
 
         int i = 0;
-        for (ColumnIdentifier id : definedOrdering.keySet())
-        {
-            ColumnIdentifier c = columnAliases.get(i);
-            if (!id.equals(c))
-            {
-                if (definedOrdering.containsKey(c))
-                    throw new InvalidRequestException(String.format("The order of columns in the CLUSTERING ORDER directive must be the one of the clustering key (%s must appear before %s)", c, id));
-                else
-                    throw new InvalidRequestException(String.format("Missing CLUSTERING ORDER for column %s", c));
+        for (auto& pair: _defined_ordering){
+            auto& id = pair.first;
+            auto& c = _column_aliases.at(i);
+
+            if (!(*id == *c)) {
+                if (find_ordering_info(c)) {
+                    throw exceptions::invalid_request_exception(sprint("The order of columns in the CLUSTERING ORDER directive must be the one of the clustering key (%s must appear before %s)", c, id));
+                } else {
+                    throw exceptions::invalid_request_exception(sprint("Missing CLUSTERING ORDER for column %s", c));
+                }
             }
             ++i;
         }
     }
-#endif
 
     return ::make_shared<parsed_statement::prepared>(stmt);
 }
@@ -343,12 +343,13 @@ data_type create_table_statement::raw_statement::get_type_and_remove(column_map_
         throw exceptions::invalid_request_exception(sprint("Invalid collection type for PRIMARY KEY component %s", t->text()));
     }
     columns.erase(t);
-#if 0
-    // FIXME: reversed types are not supported
-    Boolean isReversed = definedOrdering.get(t);
-    return isReversed != null && isReversed ? ReversedType.getInstance(type) : type;
-#endif
-    return type;
+
+    auto is_reversed = find_ordering_info(t);
+    if (!is_reversed) {
+        return type;
+    } else {
+        return *is_reversed ? reversed_type_impl::get_instance(type) : type;
+    }
 }
 
 void create_table_statement::raw_statement::add_definition(::shared_ptr<column_identifier> def, ::shared_ptr<cql3_type::raw> type, bool is_static) {
@@ -368,7 +369,7 @@ void create_table_statement::raw_statement::add_column_alias(::shared_ptr<column
 }
 
 void create_table_statement::raw_statement::set_ordering(::shared_ptr<column_identifier> alias, bool reversed) {
-    defined_ordering.emplace_back(alias, reversed);
+    _defined_ordering.emplace_back(alias, reversed);
 }
 
 void create_table_statement::raw_statement::set_compact_storage() {

cql3/statements/create_table_statement.hh

@@ -38,6 +38,7 @@
 #include <utility>
 #include <vector>
 #include <set>
+#include <experimental/optional>
 
 namespace cql3 {
 
@@ -109,7 +110,15 @@ public:
 private:
     std::vector<std::vector<::shared_ptr<column_identifier>>> _key_aliases;
     std::vector<::shared_ptr<column_identifier>> _column_aliases;
-    std::vector<std::pair<::shared_ptr<column_identifier>, bool>> defined_ordering; // Insertion ordering is important
+    std::vector<std::pair<::shared_ptr<column_identifier>, bool>> _defined_ordering; // Insertion ordering is important
+    std::experimental::optional<bool> find_ordering_info(::shared_ptr<column_identifier> type) {
+        for (auto& t: _defined_ordering) {
+            if (*(t.first) == *type) {
+                return t.second;
+            }
+        }
+        return {};
+    }
     create_table_statement::column_set_type _static_columns;
 
     bool _use_compact_storage = false;

db/marshal/type_parser.cc

@@ -132,7 +132,13 @@ data_type type_parser::get_abstract_type(const sstring& compare_with, type_parse
     } else {
         class_name = "org.apache.cassandra.db.marshal." + compare_with;
     }
-    if (class_name == "org.apache.cassandra.db.marshal.FrozenType") {
+    if (class_name == "org.apache.cassandra.db.marshal.ReversedType") {
+        auto l = parser.get_type_parameters(false);
+        if (l.size() != 1) {
+            throw exceptions::configuration_exception("ReversedType takes exactly 1 type parameter");
+        }
+        return reversed_type_impl::get_instance(l[0]);
+    } else if (class_name == "org.apache.cassandra.db.marshal.FrozenType") {
         auto l = parser.get_type_parameters(false);
         if (l.size() != 1) {
             throw exceptions::configuration_exception("FrozenType takes exactly 1 type parameter");

tests/urchin/sstable_datafile_test.cc

@@ -1349,3 +1349,65 @@ SEASTAR_TEST_CASE(datafile_generation_39) {
         }).then([sst, mtp, s] {});
     });
 }
+
+SEASTAR_TEST_CASE(datafile_generation_40) {
+    return test_setup::do_with_test_directory([] {
+        // Data file with clustering key sorted in descending order
+        //
+        // Respective CQL table and CQL insert:
+        // CREATE TABLE table (
+        //    p1 text,
+        //    c1 text,
+        //    r1 int,
+        //    PRIMARY KEY (p1, c1)
+        // ) WITH compact storage and compression = {} and clustering order by (cl1 desc);
+        // INSERT INTO table (p1, c1, r1) VALUES ('key1', 'a', 1);
+        // INSERT INTO table (p1, c1, r1) VALUES ('key1', 'b', 1);
+
+        auto s = [] {
+            schema_builder builder(make_lw_shared(schema({}, some_keyspace, some_column_family,
+                {{"p1", utf8_type}}, {{"c1", reversed_type_impl::get_instance(utf8_type)}}, {{"r1", int32_type}}, {}, utf8_type
+            )));
+            return builder.build(schema_builder::compact_storage::yes);
+        }();
+
+        auto mt = make_lw_shared<memtable>(s);
+        auto key = partition_key::from_exploded(*s, {to_bytes("key1")});
+        mutation m(key, s);
+
+        const column_definition& r1_col = *s->get_column_definition("r1");
+        auto ca = clustering_key::from_exploded(*s, {to_bytes("a")});
+        m.set_clustered_cell(ca, r1_col, make_atomic_cell(int32_type->decompose(1)));
+        mt->apply(std::move(m));
+
+        auto cb = clustering_key::from_exploded(*s, {to_bytes("b")});
+        m.set_clustered_cell(cb, r1_col, make_atomic_cell(int32_type->decompose(1)));
+        mt->apply(std::move(m));
+
+        auto sst = make_lw_shared<sstable>("tests/urchin/sstables/tests-temporary", 40, la, big);
+
+        return sst->write_components(*mt).then([mt, sst, s] {
+            auto fname = sstable::filename("tests/urchin/sstables/tests-temporary", la, 40, big, sstable::component_type::Data);
+            return engine().open_file_dma(fname, open_flags::ro).then([] (file f) {
+                auto bufptr = allocate_aligned_buffer<char>(4096, 4096);
+
+                auto fut = f.dma_read(0, bufptr.get(), 4096);
+                return std::move(fut).then([f = std::move(f), bufptr = std::move(bufptr)] (size_t size) mutable {
+                    auto buf = bufptr.get();
+                    size_t offset = 0;
+                    auto check_chunk = [buf, &offset] (std::vector<uint8_t> vec) {
+                        BOOST_REQUIRE(::memcmp(vec.data(), &buf[offset], vec.size()) == 0);
+                        offset += vec.size();
+                    };
+                    check_chunk({ /* first key */ 0, 4, 'k', 'e', 'y', '1' });
+                    check_chunk({ /* deletion time */ 0x7f, 0xff, 0xff, 0xff, 0x80, 0, 0, 0, 0, 0, 0, 0 });
+                    check_chunk({ /* first expected row name */ 0, 1, 'b' });
+                    check_chunk(/* row contents, same for both */ {/* mask */ 0, /* timestamp */ 0, 0, 0, 0, 0, 0, 0, 0, /* value */ 0, 0, 0, 4, 0, 0, 0, 1 });
+                    check_chunk({ /* second expected row name */ 0, 1, 'a' });
+                    check_chunk(/* row contents, same for both */ {/* mask */ 0, /* timestamp */ 0, 0, 0, 0, 0, 0, 0, 0, /* value */ 0, 0, 0, 4, 0, 0, 0, 1 });
+                    return f.close().finally([f] {});
+                });
+            });
+        });
+    });
+}

tests/urchin/types_test.cc

@@ -441,3 +441,67 @@ BOOST_AUTO_TEST_CASE(test_parse_recursive_type) {
     auto type = parser.parse();
     BOOST_REQUIRE(type->as_cql3_type()->to_string() == "map<int, tuple<int, int>>");
 }
+
+BOOST_AUTO_TEST_CASE(test_create_reversed_type) {
+    auto ri = reversed_type_impl::get_instance(bytes_type);
+    BOOST_REQUIRE(ri->is_reversed());
+    BOOST_REQUIRE(ri->is_value_compatible_with(*bytes_type));
+    BOOST_REQUIRE(!ri->is_compatible_with(*bytes_type));
+    auto val_lt = bytes_type->decompose(bytes("a"));
+    auto val_gt = bytes_type->decompose(bytes("b"));
+    auto straight_comp = bytes_type->compare(bytes_view(val_lt), bytes_view(val_gt));
+    auto reverse_comp = ri->compare(bytes_view(val_lt), bytes_view(val_gt));
+    BOOST_REQUIRE(straight_comp == -reverse_comp);
+}
+
+BOOST_AUTO_TEST_CASE(test_create_reverse_collection_type) {
+    auto my_set_type = set_type_impl::get_instance(bytes_type, true);
+    auto ri = reversed_type_impl::get_instance(my_set_type);
+    BOOST_REQUIRE(ri->is_reversed());
+    BOOST_REQUIRE(ri->is_collection());
+    BOOST_REQUIRE(ri->is_multi_cell());
+
+    std::vector<boost::any> first_set;
+    bytes b1("1");
+    bytes b2("2");
+    first_set.push_back(boost::any(b1));
+    first_set.push_back(boost::any(b2));
+
+    std::vector<boost::any> second_set;
+    bytes b3("2");
+    second_set.push_back(boost::any(b1));
+    second_set.push_back(boost::any(b3));
+
+    auto bv1 = my_set_type->decompose(first_set);
+    auto bv2 = my_set_type->decompose(second_set);
+
+    auto straight_comp = my_set_type->compare(bytes_view(bv1), bytes_view(bv2));
+    auto reverse_comp = ri->compare(bytes_view(bv2), bytes_view(bv2));
+    BOOST_REQUIRE(straight_comp == -reverse_comp);
+}
+
+BOOST_AUTO_TEST_CASE(test_parse_reversed_type) {
+    sstring value("org.apache.cassandra.db.marshal.ReversedType(org.apache.cassandra.db.marshal.Int32Type)");
+    auto parser = db::marshal::type_parser(value);
+    auto ri = parser.parse();
+    BOOST_REQUIRE(ri->as_cql3_type()->to_string() == "int");
+    BOOST_REQUIRE(ri->is_reversed());
+    BOOST_REQUIRE(ri->is_value_compatible_with(*int32_type));
+    BOOST_REQUIRE(!ri->is_compatible_with(*int32_type));
+
+    auto val_lt = int32_type->decompose(1);
+    auto val_gt = int32_type->decompose(2);
+    auto straight_comp = int32_type->compare(bytes_view(val_lt), bytes_view(val_gt));
+    auto reverse_comp = ri->compare(bytes_view(val_lt), bytes_view(val_gt));
+    BOOST_REQUIRE(straight_comp == -reverse_comp);
+}
+
+BOOST_AUTO_TEST_CASE(test_reversed_type_value_compatibility) {
+    auto rb = reversed_type_impl::get_instance(bytes_type);
+    auto rs = reversed_type_impl::get_instance(utf8_type);
+
+    BOOST_REQUIRE(!rb->is_compatible_with(*bytes_type));
+    BOOST_REQUIRE(!rb->is_compatible_with(*utf8_type));
+    BOOST_REQUIRE(rb->is_value_compatible_with(*rs));
+    BOOST_REQUIRE(rb->is_value_compatible_with(*utf8_type));
+}

types.hh

@@ -482,6 +482,97 @@ template <typename InternedType, typename... BaseTypes>
 thread_local typename type_interning_helper<InternedType, BaseTypes...>::map_type
     type_interning_helper<InternedType, BaseTypes...>::_instances;
 
+class reversed_type_impl : public abstract_type {
+    using intern = type_interning_helper<reversed_type_impl, data_type>;
+    friend struct shared_ptr_make_helper<reversed_type_impl, true>;
+
+    data_type _underlying_type;
+    reversed_type_impl(data_type t) : abstract_type("org.apache.cassandra.db.marshal.ReversedType(" + t->name() + ")"), _underlying_type(t) {}
+protected:
+    virtual bool is_value_compatible_with_internal(const abstract_type& other) const {
+        return _underlying_type->is_value_compatible_with(*(other.underlying_type()));
+    }
+public:
+    virtual int32_t compare(bytes_view v1, bytes_view v2) const override {
+        return _underlying_type->compare(v2, v1);
+    }
+    virtual bool less(bytes_view v1, bytes_view v2) const override {
+        return _underlying_type->less(v2, v1);
+    }
+
+    virtual bool equal(bytes_view v1, bytes_view v2) const override {
+        return _underlying_type->equal(v1, v2);
+    }
+
+    virtual void validate(bytes_view v) const override {
+        _underlying_type->validate(v);
+    }
+
+    virtual void validate_collection_member(bytes_view v, const bytes& collection_name) const  override {
+        _underlying_type->validate_collection_member(v, collection_name);
+    }
+
+    virtual bool is_compatible_with(const abstract_type& previous) const override {
+        if (previous.is_reversed()) {
+            return _underlying_type->is_compatible_with(*previous.underlying_type());
+        }
+        return false;
+    }
+
+    virtual shared_ptr<const abstract_type> underlying_type() const override {
+        return _underlying_type;
+    }
+
+    virtual bool is_byte_order_comparable() const override {
+        return _underlying_type->is_byte_order_comparable();
+    }
+    virtual bool is_byte_order_equal() const override {
+        return _underlying_type->is_byte_order_equal();
+    }
+    virtual size_t hash(bytes_view v) const override {
+        return _underlying_type->hash(v);
+    }
+    virtual bool is_reversed() const override { return true; }
+    virtual bool is_counter() const override {
+        return _underlying_type->is_counter();
+    }
+    virtual bool is_collection() const override {
+        return _underlying_type->is_collection();
+    }
+    virtual bool is_multi_cell() const override {
+        return _underlying_type->is_multi_cell();
+    }
+
+    virtual void serialize(const boost::any& value, bytes::iterator& out) const override {
+        _underlying_type->serialize(value, out);
+    }
+    virtual size_t serialized_size(const boost::any& value) const {
+        return _underlying_type->serialized_size(value);
+    }
+    virtual boost::any deserialize(bytes_view v) const override {
+        return _underlying_type->deserialize(v);
+    }
+
+    virtual sstring get_string(const bytes& b) const override {
+        return _underlying_type->get_string(b);
+    }
+    virtual sstring to_string(const bytes& b) const override {
+        return _underlying_type->to_string(b);
+    }
+    virtual bytes from_string(sstring_view s) const override {
+        return _underlying_type->from_string(s);
+    }
+
+    virtual ::shared_ptr<cql3::cql3_type> as_cql3_type() const override {
+        return _underlying_type->as_cql3_type();
+    }
+
+    static shared_ptr<const reversed_type_impl> get_instance(data_type type) {
+        return intern::get_instance(std::move(type));
+    }
+};
+using reversed_type = shared_ptr<const reversed_type_impl>;
+
 class map_type_impl final : public collection_type_impl {
     using map_type = shared_ptr<const map_type_impl>;
     using intern = type_interning_helper<map_type_impl, data_type, data_type, bool>;