wasm udf: deserialize counters as integers

Currently, because serialize_visitor::operator() is not implemented
for counters, we cannot convert a counter returned by a WASM UDF
to bytes when returning from wasm::run_script().

We could disallow using counters as WASM UDF return types, but an
easier solution which we're already using in Lua UDFs is treating
the returned counters as 64-bit integers when deserializing. This
patch implements the latter approach and adds a test for it.

lang/wasm.cc

@@ -178,7 +178,13 @@ struct from_val_visitor {
         if (ret_size == -1) {
             return bytes_opt{};
         }
-        bytes_opt ret = t.decompose(t.deserialize(bytes_view(reinterpret_cast<int8_t*>(data), ret_size)));
+        bytes_opt ret;
+        const counter_type_impl* counter = dynamic_cast<const counter_type_impl*>(&t);
+        if (counter) {
+            ret = long_type->decompose(t.deserialize(bytes_view(reinterpret_cast<int8_t*>(data), ret_size)));
+        } else {
+            ret = t.decompose(t.deserialize(bytes_view(reinterpret_cast<int8_t*>(data), ret_size)));
+        }
 
         if (wasmtime::get_abi(instance, store, *memory) == 2) {
             auto free_func = wasmtime::create_func(instance, store, "_scylla_free");

test/cql-pytest/test_wasm.py

@@ -1094,5 +1094,16 @@ def test_drop(cql, test_keyspace, table1, scylla_with_wasm_only, metrics):
             assert(get_metric(metrics, 'scylla_user_functions_cache_instace_count_any') > 0)
         assert(get_metric(metrics, 'scylla_user_functions_cache_instace_count_any') == 0)
 
-
-
+# Test that we can use counters as the return type of a WASM UDF.
+def test_counter(cql, test_keyspace, scylla_only):
+    schema = "p int, c counter, PRIMARY KEY (p)"
+    ri_counter_name = unique_name()
+    ri_counter_source = read_function_from_file('return_input', ri_counter_name)
+    src = f"(input counter) RETURNS NULL ON NULL INPUT RETURNS counter LANGUAGE xwasm AS '{ri_counter_source}'"
+    with new_test_table(cql, test_keyspace, schema) as table:
+        cql.execute(f"UPDATE {table} SET c = c + 2  WHERE p = 42;")
+        with new_function(cql, test_keyspace, src, ri_counter_name):
+            assert cql.execute(f"SELECT {ri_counter_name}(c) AS result FROM {table} WHERE p = 42").one().result == 2
+            cql.execute(f"UPDATE {table} SET c = c + 1  WHERE p = 42;")
+            cql.execute(f"UPDATE {table} SET c = c - 4  WHERE p = 42;")
+            assert cql.execute(f"SELECT {ri_counter_name}(c) AS result FROM {table} WHERE p = 42").one().result == -1