wait_for_schema_agreement() too slow?

[mykaul]

Resuming a discussion from @nyh and @Lorak-mmk from scylladb/scylladb#23138 on why schema agreement may be slow (and for the sake of discussion, let's also assume we are dealing with a multiple AZ, multiple DC, large cluster, say 100 nodes):

1. There's a 0.2 seconds sleep between cycles of trying to get an agreed upon schema -

   python-driver/cassandra/cluster.py

   Line 4268 in a401409

   self._time.sleep(0.2)

   - is that a real issue?

2. python-driver/cassandra/cluster.py

   Line 4275 in a401409

   def _get_schema_mismatches(self, peers_result, local_result, local_address):

   maybe is an overkill - creating an populating a dictionary from all peers, just to see if there's a mismatch? Not sure it's a real issue, but clearly could be done better
3. Is the query that _get_peers_query() is using ( see

   python-driver/cassandra/cluster.py

   Line 4299 in a401409

   def _get_peers_query(self, peers_query_type, connection=None):

   ) optimal for this use case? What do we need other than the schema_version ? Seem there are 3 options:

_SELECT_PEERS = "SELECT * FROM system.peers"
_SELECT_PEERS_NO_TOKENS_TEMPLATE = "SELECT host_id, peer, data_center, rack, rpc_address, {nt_col_name}, release_version, schema_version FROM system.peers"
_SELECT_SCHEMA_PEERS_TEMPLATE = "SELECT peer, host_id, {nt_col_name}, schema_version FROM system.peers"

I think (just from reading the code) we use _SELECT_SCHEMA_PEERS_TEMPLATE , which seems reasonable.

Anything else? Do we even have an issue?

[nyh]

@mykaul the reason I didn't open this issue myself is that I don't actually know that it exists. Do we actually know that the around-one-second table creation time reported in scylladb/scylladb#23138 is caused by waiting for schema agreement? I suggested in this issue how to disable wait-for-schema-agreement in the test to check if this is really the issue.

By the way, if I had to guess without checking, my guess is that the 0.2 second sleep does explain the long test times. Remember that what was measured wasn't just creating one table - it was actually four schema operations - create keyspace, create table, and then delete table and delete keyspace. These are four schema operations, and each of them waits for schema agreement separately.

[mykaul]

@mykaul the reason I didn't open this issue myself is that I don't actually know that it exists. Do we actually know that the around-one-second table creation time reported in scylladb/scylladb#23138 is caused by waiting for schema agreement? I suggested in this issue how to disable wait-for-schema-agreement in the test to check if this is really the issue.

I don't know if it exists either, but I did not want to pollute the other PR (I'm also bad at this and do it too often), did not want to neglect the issue and certainly did not want to disable wait-for-schema-agreement, which cannot be a good thing.

By the way, if I had to guess without checking, my guess is that the 0.2 second sleep does explain the long test times. Remember that what was measured wasn't just creating one table - it was actually four schema operations - create keyspace, create table, and then delete table and delete keyspace. These are four schema operations, and each of them waits for schema agreement separately.

I agree, but I also doubt if, when we have 1000 clients connected to the cluster, we want them to hit the cluster with these queries too much on a schema change? So from that perspective, 0.2 is not that bad.
Anyway, for this discussion, I've opened the issue. Perhaps it's not a real issue and we can move on!

[Lorak-mmk]

1. There's a 0.2 seconds sleep between cycles of trying to get an agreed upon schema - 
    - is that a real issue?

I don't think the sleep itself is an issue. The alternative is issuing requests without any sleep, which would be bad for the cluster.
What we could change is make this interval configurable - it is configurable in Rust Driver (with a default of 200ms as here).
Do we have good reason to believe it would help with Scylla tests? I think we should check that first.

2. maybe is an overkill - creating an populating a dictionary from all peers, just to see if there's a mismatch? Not sure it's a real issue, but clearly could be done better

I'd be very surprised if changing this brought any measurable improvement.

3. Is the query that _get_peers_query() is using optimal for this use case? What do we need other than the schema_version ? Seem there are 3 options:

_SELECT_PEERS = "SELECT * FROM system.peers"
_SELECT_PEERS_NO_TOKENS_TEMPLATE = "SELECT host_id, peer, data_center, rack, rpc_address, {nt_col_name}, release_version, schema_version FROM system.peers"
_SELECT_SCHEMA_PEERS_TEMPLATE = "SELECT peer, host_id, {nt_col_name}, schema_version FROM system.peers"

I think (just from reading the code) we use _SELECT_SCHEMA_PEERS_TEMPLATE , which seems reasonable.

I agree that it looks like we use this one, and it seems reasonable.

Anything else? Do we even have an issue?

I wonder what should be the preferred method for checking schema agreement.

Python takes a single connection (when awaiting agreement after issuing DDL it will be the connection that the request was sent on, when handling events etc it will be control connection), and uses this connection to query system.local and system.peers, then checks if all returned schema version are the same.

Java Driver 3.x seems to also use a single connection and query both tables. And like Python Driver it doesn't select too much stuff:

  private static final String SELECT_SCHEMA_PEERS =
      "SELECT peer, rpc_address, schema_version, host_id FROM system.peers";
  private static final String SELECT_SCHEMA_LOCAL =
      "SELECT schema_version, host_id FROM system.local WHERE key='local'";

Java Driver 4.x also uses both tables, but selects everything from system.peers - this could potentially be improved:

      CompletionStage<AdminResult> localQuery =
          query("SELECT schema_version FROM system.local WHERE key='local'");
      CompletionStage<AdminResult> peersQuery = query("SELECT * FROM system.peers");

GoCQL does the same, and also selects too much stuff.

Rust Driver is (probably) the only one that does it differently.
It takes all working connections (we could probably change that to all nodes. Schema version is always the same between shards, right? ). and selects schema_version from system.local on each connection. Then it checks if all results are the same.
I see it was introduced here: scylladb/scylla-rust-driver#250 and I'm wondering what race it is talking about - it is not clear from the PR message. @piodul - you reviewed it so maybe you know?

This solution issues more queries, but only small ones, to system.local. It does not touch system.peers at all. Is there more pressure on the cluster because of that? I have no idea, maybe. We'd need some benchmarks to say.
In scylladb/scylladb#23138 one advantage was discovered - it does not rely on gossip, only Raft (because system.peers is updated using gossip).

Another advantage is ease of use (from the perspective of the developer of the driver). When using system.peers solution you must take care to check agreement on the right connection. With the Rust Driver solution you don't need to worry about such stuff at all.
For example, if you check it after DDL, and you use a different connection than the one you issues DDL on, then this other node may not even know about this yet, and agreement may return false positive.
Maybe the above is not longer true with Raft, I don't know. But it should still be true for Cassandra anyway.

Which approach is better?

[mykaul]

Thanks @Lorak-mmk for looking into it - I think you summarized it well and it seems like we just need to improve the Java driver's query to system.peers - I would not touch anything else, personally.

I think the Rust way of checking is interesting, seems a bit of an overkill these days (with Raft).

[dawmd]

I don't know if this is relevant, but sharing nonetheless: I ran into this line today:

https://github.com/scylladb/scylladb/blob/cd61f6054955f41325d1a46a4f07c43041f94799/cql3/query_processor.cc#L1072

[Lorak-mmk]

I don't know if this is relevant, but sharing nonetheless: I ran into this line today:

https://github.com/scylladb/scylladb/blob/cd61f6054955f41325d1a46a4f07c43041f94799/cql3/query_processor.cc#L1072

It doesn't say how it wants to eliminate the timeouts, so not really helpful.
Please note that you can disable automatic schema agreement wait after issuing a schema-changing query, by setting max_schema_agreement_wait to 0.

Thanks @Lorak-mmk for looking into it - I think you summarized it well and it seems like we just need to improve the Java driver's query to system.peers - I would not touch anything else, personally.

I think the Rust way of checking is interesting, seems a bit of an overkill these days (with Raft).

I opened the issues about this in Java Driver and gocql:

• 4.x: Don't use overly broad SELECTs when checking schema agreement java-driver#468
• Overly broad SELECT when awaiting schema agreement. gocql#408

So I think this one can be closed @mykaul

[piodul]

Rust Driver is (probably) the only one that does it differently. It takes all working connections (we could probably change that to all nodes. Schema version is always the same between shards, right? ). and selects schema_version from system.local on each connection. Then it checks if all results are the same. I see it was introduced here: scylladb/scylla-rust-driver#250 and I'm wondering what race it is talking about - it is not clear from the PR message. @piodul - you reviewed it so maybe you know?

I'm not sure, it was four years ago.

Judging by the description of the issue that the PR fixed, I expect that the problem was as follows: after you issue a schema-altering statement to one of the nodes, it updates its local schema. However, session.await_schema_agreement could choose a different node, fetch its system.local and system.peers and then decide whether the schema is in sync or not. If these queries are sent to a different node than the one that was used for the schema-altering statement, you don't have a guarantee that the node learned about the new schema at all - the tables could report that the schema is in sync, with the previous schema version.

[mykaul]

Why, with Raft, you need to wait for schema agreement anyway? If a node is reached that did not learn about the new schema, isn't it the node's work to fetch the new schema?

[Lorak-mmk]

Why, with Raft, you need to wait for schema agreement anyway? If a node is reached that did not learn about the new schema, isn't it the node's work to fetch the new schema?

Well there is Cassandra - but we can of course add some if scylla conditions, so I'll assume that is what we are talking about.

I do not know what guarantees our Raft implementation has. Is it documented somewhere? Specifically:

• If I send DDL to some node, when it send me response? After all nodes learn new schema? After just this node learns new schema?
• Is it guaranteed that if I send the DDL (let's say CREATE TABLE) to node X, receive a response, then send INSERT INTO new_table to node Y then Y will know about this table and execute the request?

[nyh]

Why, with Raft, you need to wait for schema agreement anyway? If a node is reached that did not learn about the new schema, isn't it the node's work to fetch the new schema?

I think the idea is the following:
Imagine you execute a CREATE TABLE x command. The driver decides to send it to node A.
Now, you execute a INSERT INTO x ... command. The driver decides to send it to a different node - B.
But what if the node B didn't yet learn about x's existance?
We definitely do not attempt to pull a schema for a table "x" just because it seems it doesn't exist.

So the the solution we (and Cassandra) have right now in the drivers is:

1. Scylla returns a "schema updated" flag when it completes a CREATE TABLE command
2. When the driver sees this flag, it "waits for schema agreement" - i.e., it waits for all nodes, not just A, to have the same schema.
3. After this wait, it is safe to send any command, including, INSERT INTO x, to any node.

I don't understand how Raft makes this wait unnecessary. Maybe it needs to be done differently, but the wait is still necessary. It just shouldn't take 1.0 seconds (or 0.2 seconds) in the tests, we should learn almost immediately that it happened.

I am still guessing that the only problem is an explicit sleep of 0.2 seconds or something similar - and not any missing Raft support in the driver.

[mykaul]

@nyh - does the same apply for ALTER commands that perform schema changes (updates)? Doesn't that node need to pull a new schema version? (though I'm aware of scylladb/scylladb#20860 so not everything is as expected).

[nyh]

@mykaul yes, both CREATE and ALTER commands return a cql_transport::event::schema_change value. The former returns event_t::change_type::CREATED and the latter event::schema_change::change_type::UPDATED. I don't remember if the driver only needs to wait for schema agreement for the case of CREATED. My guess is that it also needs to wait in the case of UPDATED (i.e., ALTER command). Imagine that you perform an ALTER to add a column to some table, and then perform an INSERT to it with this column.

Again, surely the server will eventually need to "pull" (or use Raft or whatever) the most recent schema on all nodes. But the point discussed here is that the driver needs to wait until this happens, because there are some operations (like write to a new table or a new column) that we can't send to a node that hasn't seen this change yet.

[mykaul]

I must say that from reading the code (Python), it does seem to do a lot on schema refresh, but perhaps it's rare enough and it's only on update events.
I wonder why we (1) need all of those and (2) if they can be prepared (especially as we expect them for a table we use):

_SELECT_TABLES = "SELECT * FROM system_schema.tables"
    _SELECT_COLUMNS = "SELECT * FROM system_schema.columns"
    _SELECT_INDEXES = "SELECT * FROM system_schema.indexes"
    _SELECT_TRIGGERS = "SELECT * FROM system_schema.triggers"
    

They all have this WHERE added to them - " WHERE keyspace_name = %%s AND %s = %%s" % (self._table_name_col), (keyspace, table), _encoder)

Seems like not a big deal, but when we have 1000's of clients hitting this (even though I assume the results are cached). Do we really need to refresh all of this?
(this is just from reading the code, did not snoop traffic for real. Maybe I should!)

[nyh]

Seems like not a big deal, but when we have 1000's of clients hitting this (even though I assume the results are cached). Do we really need to refresh all of this? (this is just from reading the code, did not snoop traffic for real. Maybe I should!)

I think the "schema refresh" issue you mention now - where the driver reads the schema from the server - is separate from the "wait for schema agreement" which we discussed previously. I agree there are probably problems when 1000s of clients are doing this, and I think we saw this problem in the past when thousands of clients boot and read the schema, I don't know what we ever did about this. But this is not related to the problem we had in the tests, where there is a single cluster, and refereshing the schema on a multi-node cluster shouldn't be slower than on a one-node cluster where I know it takes only a few milliseconds at worst (so single-node tests don't complain about extremely slow table creation).