cluster: compute_token() shares existing logic

The recently added `calculate_token_for_partition_key()` duplicates the
logic that `ClusterData::compute_token()` uses. To deduplicate,
`compute_token()` now calls `calculate_token_for_partition_key()`.

scylla/src/transport/cluster.rs

@@ -1,8 +1,10 @@
 /// Cluster manages up to date information and connections to database nodes
 use crate::frame::response::event::{Event, StatusChangeEvent};
 use crate::frame::value::ValueList;
+use crate::prepared_statement::PartitionKeyError;
 use crate::routing::Token;
 use crate::transport::host_filter::HostFilter;
+use crate::transport::partitioner::calculate_token_for_partition_key;
 use crate::transport::{
     connection::{Connection, VerifiedKeyspaceName},
     connection_pool::PoolConfig,
@@ -35,7 +37,6 @@ pub struct ContactPoint {
 }
 
 use super::locator::ReplicaLocator;
-use super::partitioner::{Partitioner, PartitionerHasher};
 use super::topology::Strategy;
 
 /// Cluster manages up to date information and connections to database nodes.
@@ -409,27 +410,14 @@ impl ClusterData {
             .and_then(|t| t.partitioner.as_deref())
             .and_then(PartitionerName::from_str)
             .unwrap_or_default();
-        let mut partitioner_hasher = partitioner.build_hasher();
-        let serialized_values = partition_key.serialized()?;
-        // Null values are skipped in computation; null values in partition key are unsound,
-        // but it is consistent with computation of prepared statements token.
-        match serialized_values.len() {
-            0 => (),
-            1 => partitioner_hasher
-                .write(serialized_values.iter().next().unwrap().unwrap_or_default()),
-            _ => {
-                for value in serialized_values.iter().flatten() {
-                    let value_size: u16 = value
-                        .len()
-                        .try_into()
-                        .map_err(|_| BadQuery::ValuesTooLongForKey(value.len(), u16::MAX.into()))?;
-                    partitioner_hasher.write(&value_size.to_be_bytes());
-                    partitioner_hasher.write(value);
-                    partitioner_hasher.write(&[0u8]);
+
+        calculate_token_for_partition_key(&partition_key.serialized().unwrap(), &partitioner)
+            .map_err(|err| match err {
+                PartitionKeyError::NoPkIndexValue(_, _) => unreachable!(), // calculate_token_for_partition_key() never returns this variant
+                PartitionKeyError::ValueTooLong(values_len) => {
+                    BadQuery::ValuesTooLongForKey(values_len, u16::MAX.into())
                 }
-            }
-        };
-        Ok(partitioner_hasher.finish())
+            })
     }
 
     /// Access to replicas owning a given token