Make counters work with CDC

[jul-stas]

A table with counters cannot have CDC enabled:

cqlsh> CREATE TABLE jul.tb1 (id int PRIMARY KEY, cnt counter) with cdc={'enabled':true};
ConfigurationException: Cannot add a non counter column (cdc$deleted_cnt) in a counter column family

This happens for a purely technical reason: counter columns are territorial animals and don't allow anyone else into CF, apart from PK and CK (and other counters). If we try to enable CDC for a table-with-counters, CDC tries to create "ordinary" columns like ttl, cdc$deleted* which are in conflict with counter column(s). The CDC column type (here: counter_type) is inferred from the base table, hence the error.

I have a quickpatch with sum-of-local-counter-shards being cast to bigint, but this would not be reliable - please imagine a situation where two counter-leaders concurrently update their own counter-shards - what CDC should both of them log? Should they reconcile their CDC writes? I don't know, my proposals are:

1. do CDC logging per counter-shard. Every row of CDC "delta" log would have the counter-update and shard ID;
2. keep the track of (shard, update) in a single CDC row, in unfrozen map. Merge the updates within... idk, some time frame?

I'm not sure that clients want to be bothered with "counter shards". I'm even less sure about reconciliation of CDC logs for counter writes.
Preimage/postimage? even more issues may arise.

Please, comment @avikivity @slivne @haaawk @elcallio @kbr- @gleb-cloudius

[haaawk]

I think that representing counter as a map<shard_id, int64> would be best. This way we would be able to replicate counter value to another Scylla instance. This map should be non-frozen and each counter shard leader should write only updates to its own shard.

[kbr-]

Is it possible to reliably log the actual delta (difference between counter values caused by an update)?
Also, there's the third option:

3. don't allow CDC on counter tables

[haaawk]

Is it possible to reliably log the actual delta (difference between counter values caused by an update)?
Also, there's the third option:

1. don't allow CDC on counter tables

Let's first explore other options.

[jul-stas]

Is it possible to reliably log the actual delta (difference between counter values caused by an update)?

Seems doable to me - we can try augmenting counter update mutations. The cdc-deltas would be propagated alongside corresponding counter-shards-mutations, as reliably as the latter do it now..?

ATM I'm investigating how C* solved it (they have something). In the meantime we can think on how to do it perfectly.

[slivne]

I don't think its a good idea to log the delta - AFAIK thats how counters in Cassandra started and they switched to the current method.

Counters AFIAK do enforce RMW for their operation - which means you have a pre-image fetched and a post-image computed as well (since this is what we write at the end to the sstables/memtables).

Given that counters enforce a RMW for their operation - I think we should log in their case the full value always (even for the change) as a map<share,value> instead of only the delta (we do have a CQL extension to allow writing this as well).

[haaawk]

It seems that we can move forward with map<shard_id, value> idea, no?
Any findings in Cassandra, @jul-stas?

[jul-stas]

It seems that we can move forward with map<shard_id, value> idea, no?
Any findings in Cassandra, @jul-stas?

Just some indirect evidence.
I tried two tools that read C*'s CDC, but they weren't able to deserialize counter updates. So the log of a counter column is not just an integer. I also found this in the release notes:

we grab a (local) lock for each counter being updated, read the current value, then write the incremented value (not the increment itself as in C* 2.0) to the commitlog and to the memtable

Given that C*'s CDC has the logic of commitlog, we can speculate that the local counter-shards are being logged. So the map approach looks like the way to go.

[jul-stas]

I'm still unsure if it plays well with existing code...

• Imagine a table with 2 counter columns.
• Now update affects only 1 column. One counter-shards-mutation arrives in augment_..., then it's converted to 1 map and goes to CDC "delta" log - all good.
• But we want pre-/postimage to display all columns (2 counters). It's a bit clumsy because:

a. preimage_select receives pirow with summed up counter-shards. That's how select works - it gets integers, not counters. So some CDC code will have to be injected into counters internals.

b. Injected? Computing counter-shards on leader is integrated with applying this change to DB. Docs even say that this happens under a lock. That means we'll have to fetch pirow before processing counter update.

What can we do?

• in pre-/postimage display the total values of counters. So, a map with 1 entry, with some fake shard-ID vs. value of summed shards. Working prototype here: https://github.com/scylladb/scylla/compare/next...jul-stas:6013-counters-in-cdc?expand=1
  But... why display shards in "deltas" if pre-/postimage don't do that? I don't like it.
• otherwise (?) lower expectations for pre-/postimage for counters.
• do steps a.-b. anyway. These are only technical obstacles, just worrying me.