Merge 'alternator: rudimentary implementation of TTL expiration servi…

…ce' from Nadav Har'El

In this patch series we add an implementation of an
expiration service to Alternator, which periodically scans the data in
the table, looking for expired items and deleting them.

We also continue to improve the TTL test suite to cover additional
corner cases discovered during the development of the code.

This implementation is good enough to make all existing tests but one,
plus a few new ones, pass, but is still a very partial and inefficient
implementation littered with FIXMEs throughout the code. Among other
things, this initial implementation doesn't do anything reasonable about pacing of
the scan or about multiple tables, it scans entire items instead of only the
needed parts, and because each shard "owns" a different subset of the
token ranges, if a node goes down, partitions which it "owns" will not
get expired.

The current tests cannot expose these problems, so we will need to develop
additional tests for them.

Because this implementation is very partial, the Alternator TTL continues
to remain "experimental", cannot be used without explicitly enabling this
experimental feature, and must not be used for any important deployment.

Refs #5060 but doesn't close the issue (let's not close it until we have a
reasonably complete implementation - not this partial one).

Closes #9624

* github.com:scylladb/scylla:
  alternator: fix TTL expiration scanner's handling of floating point
  test/alternator: add TTL test for more data
  test/alternator: remove "xfail" tag from passing tests in test_ttl.py
  test/alternator: make test_ttl.py tests fast on Alternator
  alternator: initial implmentation of TTL expiration service
  alternator: add another unwrap_number() variant
  alternator: add find_tag() function
  test/alternator: test another corner case of TTL setting
  test/alternator: test TTL expiration for table with sort key
  test/alternator: improve basic test for TTL expiration
  test/alternator: extract is_aws() function

alternator/executor.cc

@@ -608,6 +608,25 @@ const std::map<sstring, sstring>& get_tags_of_table(schema_ptr schema) {
     return tags_extension->tags();
 }
 
+// find_tag() returns the value of a specific tag, or nothing if it doesn't
+// exist. Unlike get_tags_of_table() above, if the table is missing the
+// tags extension (e.g., is not an Alternator table) it's not an error -
+// we return nothing, as in the case that tags exist but not this tag.
+std::optional<std::string> find_tag(const schema& s, const sstring& tag) {
+    auto it1 = s.extensions().find(tags_extension::NAME);
+    if (it1 == s.extensions().end()) {
+        return std::nullopt;
+    }
+    const std::map<sstring, sstring>& tags_map =
+        static_pointer_cast<alternator::tags_extension>(it1->second)->tags();
+    auto it2 = tags_map.find(tag);
+    if (it2 == tags_map.end()) {
+        return std::nullopt;
+    } else {
+        return it2->second;
+    }
+}
+
 static bool is_legal_tag_char(char c) {
     // FIXME: According to docs, unicode strings should also be accepted.
     // Alternator currently uses a simplified ASCII approach
@@ -3331,6 +3350,7 @@ static future<executor::request_return_type> do_query(service::storage_proxy& pr
 
     auto query_state_ptr = std::make_unique<service::query_state>(client_state, trace_state, std::move(permit));
 
+    // FIXME: should be moved above, set on opts, so get_max_result_size knows it?
     command->slice.options.set<query::partition_slice::option::allow_short_read>();
     auto query_options = std::make_unique<cql3::query_options>(cl, std::vector<cql3::raw_value>{});
     query_options = std::make_unique<cql3::query_options>(std::move(query_options), std::move(paging_state));

alternator/executor.hh

@@ -85,6 +85,7 @@ class path;
 };
 
 const std::map<sstring, sstring>& get_tags_of_table(schema_ptr schema);
+std::optional<std::string> find_tag(const schema& s, const sstring& tag);
 future<> update_tags(service::migration_manager& mm, schema_ptr schema, std::map<sstring, sstring>&& tags_map);
 schema_ptr get_table(service::storage_proxy& proxy, const rjson::value& request);
 

alternator/serialization.cc

@@ -270,6 +270,21 @@ big_decimal unwrap_number(const rjson::value& v, std::string_view diagnostic) {
     }
 }
 
+std::optional<big_decimal> try_unwrap_number(const rjson::value& v) {
+    if (!v.IsObject() || v.MemberCount() != 1) {
+        return std::nullopt;
+    }
+    auto it = v.MemberBegin();
+    if (it->name != "N" || !it->value.IsString()) {
+        return std::nullopt;
+    }
+    try {
+        return big_decimal(rjson::to_string_view(it->value));
+    } catch (const marshal_exception& e) {
+        return std::nullopt;
+    }
+}
+
 const std::pair<std::string, const rjson::value*> unwrap_set(const rjson::value& v) {
     if (!v.IsObject() || v.MemberCount() != 1) {
         return {"", nullptr};

alternator/serialization.hh

@@ -23,6 +23,7 @@
 
 #include <string>
 #include <string_view>
+#include <optional>
 #include "types.hh"
 #include "schema_fwd.hh"
 #include "keys.hh"
@@ -64,6 +65,10 @@ clustering_key ck_from_json(const rjson::value& item, schema_ptr schema);
 // raises ValidationException with diagnostic.
 big_decimal unwrap_number(const rjson::value& v, std::string_view diagnostic);
 
+// try_unwrap_number is like unwrap_number, but returns an unset optional
+// when the given v does not encode a number.
+std::optional<big_decimal> try_unwrap_number(const rjson::value& v);
+
 // Check if a given JSON object encodes a set (i.e., it is a {"SS": [...]}, or "NS", "BS"
 // and returns set's type and a pointer to that set. If the object does not encode a set,
 // returned value is {"", nullptr}