4.x: TabletMap is always empty

[scrat98]

I'm not sure maybe I'm doing something wrong, but in all use cases TabletMap is empty. I did try with different RF, single and multi node cluster - the result was the same.

According to the implementation TabletMap can only be obtained on SELECT response, therefore I'm also doing read query, but incoming payload is always empty https://github.com/scylladb/java-driver/blob/scylla-4.x/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java#L483

Also, because implementation updates the TabletMap only on the read path, it makes impossible to use correct routing for write-only applications (for example, ETL, like reading from Kafka and writing to Scylla and never read from it).

This extension adds support for sending tablet info to the drivers if the request was routed to the wrong node/shard.

And in meanwhile below

The info is send when driver asks about the information that the specific tablet contains and it is directed to the wrong node/shard

Maybe because what is "request" and "asks" means does not specified, then implementation decided to update TableMap only for the read path, but maybe the same implementation on the server side, I'm not sure

Relates to: #497

Example:

  val session = CqlSessionBuilder()
    .addContactPoints(listOf(InetSocketAddress("127.0.1.1", 9042)))
    .withLocalDatacenter("datacenter1")
    .build()
  session.execute(
    """
      CREATE KEYSPACE IF NOT EXISTS my_keyspace
      WITH replication = { 'class': 'NetworkTopologyStrategy', 'replication_factor': 1 }
      AND tablets = { 'enabled': true }
      AND durable_writes = true;
    """
  )
  session.execute(
    """
      CREATE TABLE IF NOT EXISTS my_keyspace.my_table
      (
        p      text,
        c      text,
        v      text,
        PRIMARY KEY (p, c)
      ) WITH CLUSTERING ORDER BY (c ASC)
        AND compaction = { 'class' : 'SizeTieredCompactionStrategy' };
    """
  )
  session.execute(
    """
      INSERT INTO my_keyspace.my_table(p,c,v) VALUES ('hello','world','!');
    """
  )
  session.execute(
    """
      SELECT * FROM my_keyspace.my_table;
    """
  )
  val tablets = session.metadata.tabletMap.mapping
  require(tablets.isNotEmpty())

[dkropachev]

Range queries are not getting tablets information with them, add WHERE clause and it would work.
There is also a problem in this driver that does not allow it to learn tablets from inserts.
Also use prepared statements.

[scrat98]

okay, got it. But don't you think the TabletMap also must be updated for write path as well?

Since it also depend on the server implementation and on the protocol specification I would like to ask @avikivity and @mykaul if the TabletInfo should be send to the drivers also on the write path? (for write-only applications, for example, ETL, like reading from Kafka and writing to Scylla and never read from it). And if so, maybe other scylla drivers also suffer from that problem.

https://github.com/scylladb/scylladb/blob/master/docs/dev/protocol-extensions.md#sending-tablet-info-to-the-drivers

Example:

  val session = CqlSessionBuilder()
    .addContactPoints(listOf(InetSocketAddress("172.17.0.2", 9042)))
    .withLocalDatacenter("datacenter1")
    .build()
  session.execute("DROP KEYSPACE IF EXISTS my_keyspace")
  session.execute(
    """
      CREATE KEYSPACE IF NOT EXISTS my_keyspace
      WITH replication = { 'class': 'NetworkTopologyStrategy', 'replication_factor': 1 }
      AND tablets = { 'enabled': true }
      AND durable_writes = true;
    """
  )
  session.execute(
    """
      CREATE TABLE IF NOT EXISTS my_keyspace.my_table
      (
        p      text,
        c      text,
        v      text,
        PRIMARY KEY (p, c)
      ) WITH CLUSTERING ORDER BY (c ASC)
        AND compaction = { 'class' : 'SizeTieredCompactionStrategy' };
    """
  )
  val writeStatement = session.prepare("INSERT INTO my_keyspace.my_table (p, c, v) VALUES (?, ?, ?)")
  val readStatement = session.prepare("SELECT p, c, v FROM my_keyspace.my_table WHERE p = ?")
  val loadBalancingPolicy = session.context.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME)
  // just write some random data
  repeat(10_000) {
    val toWrite = writeStatement.boundStatementBuilder()
      .setString(0, UUID.randomUUID().toString())
      .setString(1, UUID.randomUUID().toString())
      .setString(2, UUID.randomUUID().toString())
      .build()
    // TabletMap is always empty and load balancing not working
    println("TabletMap: ${session.metadata.tabletMap.mapping}")
    session.execute(toWrite)
  }

Example output:

TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
....

[dkropachev]

okay, got it. But don't you think the TabletMap also must be updated for write path as well?

Since it also depend on the server implementation and on the protocol specification I would like to ask @avikivity and @mykaul if the TabletInfo should be send to the drivers also on the write path? (for write-only applications, for example, ETL, like reading from Kafka and writing to Scylla and never read from it). And if so, maybe other scylla drivers also suffer from that problem.

https://github.com/scylladb/scylladb/blob/master/docs/dev/protocol-extensions.md#sending-tablet-info-to-the-drivers

Example:

val session = CqlSessionBuilder()
.addContactPoints(listOf(InetSocketAddress("172.17.0.2", 9042)))
.withLocalDatacenter("datacenter1")
.build()
session.execute("DROP KEYSPACE IF EXISTS my_keyspace")
session.execute(
"""
CREATE KEYSPACE IF NOT EXISTS my_keyspace
WITH replication = { 'class': 'NetworkTopologyStrategy', 'replication_factor': 1 }
AND tablets = { 'enabled': true }
AND durable_writes = true;
"""
)
session.execute(
"""
CREATE TABLE IF NOT EXISTS my_keyspace.my_table
(
p text,
c text,
v text,
PRIMARY KEY (p, c)
) WITH CLUSTERING ORDER BY (c ASC)
AND compaction = { 'class' : 'SizeTieredCompactionStrategy' };
"""
)
val writeStatement = session.prepare("INSERT INTO my_keyspace.my_table (p, c, v) VALUES (?, ?, ?)")
val readStatement = session.prepare("SELECT p, c, v FROM my_keyspace.my_table WHERE p = ?")
val loadBalancingPolicy = session.context.getLoadBalancingPolicy(DriverExecutionProfile.DEFAULT_NAME)
// just write some random data
repeat(10_000) {
val toWrite = writeStatement.boundStatementBuilder()
.setString(0, UUID.randomUUID().toString())
.setString(1, UUID.randomUUID().toString())
.setString(2, UUID.randomUUID().toString())
.build()
// TabletMap is always empty and load balancing not working
println("TabletMap: ${session.metadata.tabletMap.mapping}")
session.execute(toWrite)
}
Example output:

TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
TabletMap: {}
....

Thanks, I confirm that there is problem with learning tablets info on writes in this driver, fix in progress.

[Lorak-mmk]

Maybe because what is "request" and "asks" means does not specified, then implementation decided to update TableMap only for the read path, but maybe the same implementation on the server side, I'm not sure

Since it also depend on the server implementation and on the protocol specification I would like to ask @avikivity and @mykaul if the TabletInfo should be send to the drivers also on the write path? (for write-only applications, for example, ETL, like reading from Kafka and writing to Scylla and never read from it). And if so, maybe other scylla drivers also suffer from that problem.

Afaik the server sends the tablet info back for requests (request is any kind of request, so it doesn't matter if it is a select or insert) if the request is routable (= it is possible to calculate its partition key), but was send to a coordinator that is not a replica.
What requests are not routable?

• Unprepared statements
• Requests that are related to more than one partition key (e.g. range scans as mentioned by @dkropachev )
• Requests that have some partition key values hardcoded, e.g INSERT INTO t (a, b, c) VALUES (1, ?, ?) if partition key consists of a`` and b`. Why? Driver needs to calculate hash of pk columns, but can only use bound variables for that.

So the problem is with the driver. I'm quite sure INSERT works - we use it in tablets tests of Rust Driver.

Given that both this issue and #497 exist this is a strong signal that tablet tests in this driver are quite lacking...

[avikivity]

Yes the write path should notify the driver about tablet misses. How did we overlook that? Bad bug.

[avikivity]

Oh, @Lorak-mmk says INSERT does work.

[scrat98]

@Lorak-mmk Thanks for the clarification!
I still believe it's worth validating this issue across all Scylla drivers - it shouldn't take too much time.
Since the Rust driver works as you mentioned and already has test coverage for the Tablets functionality, that’s great - less effort needed there.

[scrat98]

@Lorak-mmk But I also have concern about queries that are related to more than one partition key, I think for them scylla server should also return tablets info.

There is always a tradeoff between throughput vs latency. And I understand that Scylla decided to focus on the low latency and it seems that's why decided to do not return Tablet Info on the "requests that are related to more than one partition key", but then it means it's not good enough for high throughput applications. Let me explain.

Data model:

CREATE TABLE data(
   device_id TEXT,
   sensor_id TEXT,
   time TIMESTAMP,
   value BLOB,
   PRIMARY KEY((device_id, sensor_id), time)
)

1. Node aware batch write

Let's say we're receiving 1M sensor points per second. For example, 10k devices that're sending micro-batches of 100 sensor points every second. In reality number of devices can be far and far more, so it's even a simple example.

If we'll do it in simple way, then we'll have 1M network OPS, that very huge amount and will consume lot's of CPU on the client side (and on server side I guess) and we will waste our resources on the request/response round-trips.

To mitigate this problem we can batch statements by node to reduce number of socket writes. But it also means that BatchStatement will contain Statements with different partition key, but all of these statements belong to the same replica node. Then we can reduce our number of requests from the 1M to the 10k * number of nodes (if assume we are writing every micro-batch separately), that's much and much less. I'm using word "node" for simplicity, but you can also assume that we're grouping statements by shard.

KairosDB is doing this: https://github.com/kairosdb/kairosdb/blob/develop/src/main/java/org/kairosdb/datastore/cassandra/CQLBatch.java#L156

Suggested shard aware batches: scylladb/scylla-rust-driver#738
Relates to this issue: #345

2. Node aware batch read

Now, let's say we wanted to have latest value (with the highest timestamp) for all sensors within device. Let's say we have 10k sensors within device.

If we'll do it in simple way we need 10k requests. But we can again group our partition keys by node (or shard), to make the following:

SELECT * FROM data 
WHERE deviceId = 'mydevice' AND sensor_id IN (‘s1’, ‘s2’, ‘s3’) 
PER PARTITION LIMIT 1

If we assume we're grouping by 100 keys (the default configurable limitation on scylla server. see max_partition_key_restrictions_per_query), then we'll have only 100 requests that will avoid small request/response round-trip. But again, here is a query with multiple partition keys.

JanusGraph is doing this: https://forum.scylladb.com/t/how-to-get-node-sharding-information-using-scylla-java-driver/696

Question: am I understood correctly that for the use cases above Tablets info won't be returned by Scylla server?

[dkropachev]

Issue in this driver: #500

[dkropachev]

#500 and #497 are fixed, we are working on releasing the driver, new version should work properly.

Added issue to document tablets behavior: #508

This issue appears to be resolved

[dkropachev]

@Lorak-mmk But I also have concern about queries that are related to more than one partition key, I think for them scylla server should also return tablets info.

There is always a tradeoff between throughput vs latency. And I understand that Scylla decided to focus on the low latency and it seems that's why decided to do not return Tablet Info on the "requests that are related to more than one partition key", but then it means it's not good enough for high throughput applications. Let me explain.

Data model:

CREATE TABLE data(
device_id TEXT,
sensor_id TEXT,
time TIMESTAMP,
value BLOB,
PRIMARY KEY((device_id, sensor_id), time)
)

1. Node aware batch write

Let's say we're receiving 1M sensor points per second. For example, 10k devices that're sending micro-batches of 100 sensor points every second. In reality number of devices can be far and far more, so it's even a simple example.

If we'll do it in simple way, then we'll have 1M network OPS, that very huge amount and will consume lot's of CPU on the client side (and on server side I guess) and we will waste our resources on the request/response round-trips.

To mitigate this problem we can batch statements by node to reduce number of socket writes. But it also means that BatchStatement will contain Statements with different partition key, but all of these statements belong to the same replica node. Then we can reduce our number of requests from the 1M to the 10k * number of nodes (if assume we are writing every micro-batch separately), that's much and much less. I'm using word "node" for simplicity, but you can also assume that we're grouping statements by shard.

KairosDB is doing this: https://github.com/kairosdb/kairosdb/blob/develop/src/main/java/org/kairosdb/datastore/cassandra/CQLBatch.java#L156

Suggested shard aware batches: scylladb/scylla-rust-driver#738 Relates to this issue: #345

2. Node aware batch read

Now, let's say we wanted to have latest value (with the highest timestamp) for all sensors within device. Let's say we have 10k sensors within device.

If we'll do it in simple way we need 10k requests. But we can again group our partition keys by node (or shard), to make the following:

SELECT * FROM data
WHERE deviceId = 'mydevice' AND sensor_id IN (‘s1’, ‘s2’, ‘s3’)
PER PARTITION LIMIT 1
If we assume we're grouping by 100 keys (the default configurable limitation on scylla server. see max_partition_key_restrictions_per_query), then we'll have only 100 requests that will avoid small request/response round-trip. But again, here is a query with multiple partition keys.

JanusGraph is doing this: https://forum.scylladb.com/t/how-to-get-node-sharding-information-using-scylla-java-driver/696

Question: am I understood correctly that for the use cases above Tablets info won't be returned by Scylla server?

On both cases driver will not get any tablet information.

[dkropachev]

@scrat98 , thanks for reporting it.

[Lorak-mmk]

more than one partition key, I think for them scylla server should also return tablets info.

And which of many partition keys should it use for that?

If we'll do it in simple way, then we'll have 1M network OPS, that very huge amount and will consume lot's of CPU on the client side (and on server side I guess) and we will waste our resources on the request/response round-trips.

To mitigate this problem we can batch statements by node to reduce number of socket writes.

If you are concerned about the amount of syscalls, then you can use coalescing. Rust Driver has it, I think some other drivers also do (but I don't know if all have it): https://docs.rs/scylla/latest/scylla/client/session_builder/struct.GenericSessionBuilder.html#method.write_coalescing and https://docs.rs/scylla/latest/scylla/client/session_builder/struct.GenericSessionBuilder.html#method.write_coalescing_delay

The advantage that the batch has over that is that it will have a bit smaller overhead from CQL frames etc. But we should have strong benchmark evidence that it is really impactful before we do anything about that.

If we'll do it in simple way we need 10k requests. But we can again group our partition keys by node (or shard), to make the following:

In your whole post you are working on a very simplified assumption that there is 1:1 relation between partition and node (+ shard). Remember that there are N replicas for any partition (where N is your replication factor). Even if one replica is the same between multiple partition keys, other have a high chance to be different. This is important for other mechanisms in the drivers, like retries or speculative execution.

But again, here is a query with multiple partition keys.

That would reduce would amount of bytes sent over the network, but would it reduce the work Scylla has to do? I have no idea.
@avikivity is it much easier for Scylla to handle 3 queries in the form of:

SELECT * FROM data 
WHERE deviceId = ? AND sensor_id = ?
PER PARTITION LIMIT 1

than it is to handle one like

SELECT * FROM data 
WHERE deviceId = ? AND sensor_id IN (?, ?, ?) 
PER PARTITION LIMIT 1

?

In any case what you propose regarding SELECT would require changes in both the server and the clients. I imagine there would need to be good evidence that introducing it is worth it.
This is widely beyond the scope of this issue. I invite you to open an issue in the Scylla repo to discuss a change like that.

[dkropachev]

4.19.0.1 is released, be available in 3 hours on maven

[scrat98]

@Lorak-mmk yes, for sure, I'll create another issue and before that I would like to finish my benchmark. But I'm currently using node aware batches for read and write in my application and it provides big improvements on the total throughput (in 4 times on write and in 30 times on read). In java client we don't need to do anything to make it work, that's why I was asking when the server is returning Tablets Info.

1. Yes, syscalls costs a lot in that case. And for java driver it's not possible to configure coalescing.
2. But also there are lot's of costs on each request acknowledgment (when you're writing). Using batch we acknowledge only one batch and not each statement.
3. You're absolutely right about retries and speculative execution. I'm grouping by first node from LoadBalancing.newQueryPlan and in java you can force to use specified node (via Statement.setNode) to avoid problems during batch execution, because query plan will be calculated again during execution and another node may be selected. So, I think about RF in my case.
4. Using IN statement with multiple partition keys should provide read improvements if the data belongs to the same node/shard, because we only need to scan SSTable once. Think about Secondary indexes, when first querying the Materialized View and then fetching the corresponding rows from the base table using the primary keys obtained from the result and internally it should be the same as using IN statement.