-
-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(scylla): upgrade Scylla driver and add UDT support (#37)
This commit upgrades the Scylla driver to version 0.13.1 and adds support for User Defined Types (UDTs). The `QueryParameter` struct has been updated to handle UDTs and the `QueryResult` struct now parses UDTs correctly. The `execute`, `query`, and `batch` methods in `ScyllaSession` have been updated to handle parameters of UDTs. The `Uuid` struct has been updated to be cloneable and copyable. Signed-off-by: Daniel Boll <[email protected]>
- Loading branch information
1 parent
17c7b7b
commit 662352e
Showing
7 changed files
with
208 additions
and
64 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| import { Cluster } from "../index.js"; | ||
|
|
||
| const nodes = process.env.CLUSTER_NODES?.split(",") ?? ["127.0.0.1:9042"]; | ||
|
|
||
| console.log(`Connecting to ${nodes}`); | ||
|
|
||
| const cluster = new Cluster({ nodes }); | ||
| const session = await cluster.connect(); | ||
|
|
||
| await session.execute( | ||
| "CREATE KEYSPACE IF NOT EXISTS udt WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }", | ||
| ); | ||
| await session.useKeyspace("udt"); | ||
|
|
||
| await session.execute( | ||
| "CREATE TYPE IF NOT EXISTS address (street text, neighbor text)", | ||
| ); | ||
| await session.execute( | ||
| "CREATE TABLE IF NOT EXISTS user (name text, address address, primary key (name))", | ||
| ); | ||
|
|
||
| interface User { | ||
| name: string; | ||
| address: { | ||
| street: string; | ||
| neighbor: string; | ||
| }; | ||
| } | ||
|
|
||
| const user: User = { | ||
| name: "John Doe", | ||
| address: { | ||
| street: "123 Main St", | ||
| neighbor: "Downtown", | ||
| }, | ||
| }; | ||
|
|
||
| await session.execute("INSERT INTO user (name, address) VALUES (?, ?)", [ | ||
| user.name, | ||
| user.address, | ||
| ]); | ||
|
|
||
| const users = (await session.execute("SELECT * FROM user")) as User[]; | ||
| console.log(users); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,25 +1,90 @@ | ||
| use std::collections::HashMap; | ||
|
|
||
| use crate::types::uuid::Uuid; | ||
| use napi::bindgen_prelude::Either3; | ||
| use scylla::_macro_internal::SerializedValues; | ||
| use napi::bindgen_prelude::{Either3, Either4}; | ||
| use scylla::{ | ||
| frame::response::result::CqlValue, | ||
| serialize::{ | ||
| row::{RowSerializationContext, SerializeRow}, | ||
| value::SerializeCql, | ||
| RowWriter, SerializationError, | ||
| }, | ||
| }; | ||
|
|
||
| pub struct QueryParameter { | ||
| pub(crate) parameters: Option<Vec<Either3<u32, String, Uuid>>>, | ||
| pub struct QueryParameter<'a> { | ||
| #[allow(clippy::type_complexity)] | ||
| pub(crate) parameters: | ||
| Option<Vec<Either4<u32, String, &'a Uuid, HashMap<String, Either3<u32, String, &'a Uuid>>>>>, | ||
| } | ||
|
|
||
| impl QueryParameter { | ||
| pub fn parser(parameters: Option<Vec<Either3<u32, String, &Uuid>>>) -> Option<SerializedValues> { | ||
| parameters | ||
| .map(|params| { | ||
| let mut values = SerializedValues::with_capacity(params.len()); | ||
| for param in params { | ||
| match param { | ||
| Either3::A(number) => values.add_value(&(number as i32)).unwrap(), | ||
| Either3::B(str) => values.add_value(&str).unwrap(), | ||
| Either3::C(uuid) => values.add_value(&(uuid.uuid)).unwrap(), | ||
| impl<'a> SerializeRow for QueryParameter<'a> { | ||
| fn serialize( | ||
| &self, | ||
| ctx: &RowSerializationContext<'_>, | ||
| writer: &mut RowWriter, | ||
| ) -> Result<(), SerializationError> { | ||
| if let Some(parameters) = &self.parameters { | ||
| for (i, parameter) in parameters.iter().enumerate() { | ||
| match parameter { | ||
| Either4::A(num) => { | ||
| CqlValue::Int(*num as i32) | ||
| .serialize(&ctx.columns()[i].typ, writer.make_cell_writer())?; | ||
| } | ||
| Either4::B(str) => { | ||
| CqlValue::Text(str.to_string()) | ||
| .serialize(&ctx.columns()[i].typ, writer.make_cell_writer())?; | ||
| } | ||
| Either4::C(uuid) => { | ||
| CqlValue::Uuid(uuid.get_inner()) | ||
| .serialize(&ctx.columns()[i].typ, writer.make_cell_writer())?; | ||
| } | ||
| Either4::D(map) => { | ||
| CqlValue::UserDefinedType { | ||
| // FIXME: I'm not sure why this is even necessary tho, but if it's and makes sense we'll have to make it so we get the correct info | ||
| keyspace: "keyspace".to_string(), | ||
| type_name: "type_name".to_string(), | ||
| fields: map | ||
| .iter() | ||
| .map(|(key, value)| match value { | ||
| Either3::A(num) => (key.to_string(), Some(CqlValue::Int(*num as i32))), | ||
| Either3::B(str) => (key.to_string(), Some(CqlValue::Text(str.to_string()))), | ||
| Either3::C(uuid) => (key.to_string(), Some(CqlValue::Uuid(uuid.get_inner()))), | ||
| }) | ||
| .collect::<Vec<(String, Option<CqlValue>)>>(), | ||
| } | ||
| .serialize(&ctx.columns()[i].typ, writer.make_cell_writer())?; | ||
| } | ||
| } | ||
| values | ||
| }) | ||
| .or(Some(SerializedValues::new())) | ||
| } | ||
| } | ||
| Ok(()) | ||
| } | ||
|
|
||
| fn is_empty(&self) -> bool { | ||
| self.parameters.is_none() || self.parameters.as_ref().unwrap().is_empty() | ||
| } | ||
| } | ||
|
|
||
| impl<'a> QueryParameter<'a> { | ||
| #[allow(clippy::type_complexity)] | ||
| pub fn parser( | ||
| parameters: Option< | ||
| Vec<Either4<u32, String, &'a Uuid, HashMap<String, Either3<u32, String, &'a Uuid>>>>, | ||
| >, | ||
| ) -> Option<Self> { | ||
| if parameters.is_none() { | ||
| return Some(QueryParameter { parameters: None }); | ||
| } | ||
|
|
||
| let parameters = parameters.unwrap(); | ||
|
|
||
| let mut params = Vec::with_capacity(parameters.len()); | ||
| for parameter in parameters { | ||
| params.push(parameter); | ||
| } | ||
|
|
||
| Some(QueryParameter { | ||
| parameters: Some(params), | ||
| }) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,58 +1,73 @@ | ||
| use scylla::frame::response::result::ColumnType; | ||
| use scylla::frame::response::result::{ColumnType, CqlValue}; | ||
| pub struct QueryResult { | ||
| pub(crate) result: scylla::QueryResult, | ||
| } | ||
|
|
||
| impl QueryResult { | ||
| pub fn parser(result: scylla::QueryResult) -> serde_json::Value { | ||
| if result.result_not_rows().is_ok() { | ||
| return serde_json::json!([]); | ||
| } | ||
|
|
||
| if result.rows.is_none() { | ||
| if result.result_not_rows().is_ok() || result.rows.is_none() { | ||
| return serde_json::json!([]); | ||
| } | ||
|
|
||
| let rows = result.rows.unwrap(); | ||
| let column_specs = result.col_specs; | ||
|
|
||
| let mut result = serde_json::json!([]); | ||
| let mut result_json = serde_json::json!([]); | ||
|
|
||
| for row in rows { | ||
| let mut row_object = serde_json::Map::new(); | ||
|
|
||
| for (i, column) in row.columns.iter().enumerate() { | ||
| let column_name = column_specs[i].name.clone(); | ||
|
|
||
| let column_value = match column { | ||
| Some(column) => match column_specs[i].typ { | ||
| ColumnType::Ascii => serde_json::Value::String(column.as_ascii().unwrap().to_string()), | ||
| ColumnType::Text => serde_json::Value::String(column.as_text().unwrap().to_string()), | ||
| ColumnType::Uuid => serde_json::Value::String(column.as_uuid().unwrap().to_string()), | ||
| ColumnType::Int => serde_json::Value::Number( | ||
| serde_json::Number::from_f64(column.as_int().unwrap() as f64).unwrap(), | ||
| ), | ||
| ColumnType::Float => serde_json::Value::Number( | ||
| serde_json::Number::from_f64(column.as_float().unwrap() as f64).unwrap(), | ||
| ), | ||
| ColumnType::Timestamp => { | ||
| serde_json::Value::String(column.as_date().unwrap().to_string()) | ||
| } | ||
| ColumnType::Date => serde_json::Value::String(column.as_date().unwrap().to_string()), | ||
| _ => "Not implemented".into(), | ||
| }, | ||
| None => serde_json::Value::Null, | ||
| }; | ||
|
|
||
| let column_value = Self::parse_value(column, &column_specs[i].typ); | ||
| row_object.insert(column_name, column_value); | ||
| } | ||
|
|
||
| result | ||
| result_json | ||
| .as_array_mut() | ||
| .unwrap() | ||
| .push(serde_json::Value::Object(row_object)); | ||
| } | ||
|
|
||
| result | ||
| result_json | ||
| } | ||
|
|
||
| fn parse_value(column: &Option<CqlValue>, column_type: &ColumnType) -> serde_json::Value { | ||
| match column { | ||
| Some(column) => match column_type { | ||
| ColumnType::Ascii => serde_json::Value::String(column.as_ascii().unwrap().to_string()), | ||
| ColumnType::Text => serde_json::Value::String(column.as_text().unwrap().to_string()), | ||
| ColumnType::Uuid => serde_json::Value::String(column.as_uuid().unwrap().to_string()), | ||
| ColumnType::Int => serde_json::Value::Number( | ||
| serde_json::Number::from_f64(column.as_int().unwrap() as f64).unwrap(), | ||
| ), | ||
| ColumnType::Float => serde_json::Value::Number( | ||
| serde_json::Number::from_f64(column.as_float().unwrap() as f64).unwrap(), | ||
| ), | ||
| ColumnType::Timestamp | ColumnType::Date => { | ||
| serde_json::Value::String(column.as_cql_date().unwrap().0.to_string()) | ||
| } | ||
| ColumnType::UserDefinedType { field_types, .. } => { | ||
| Self::parse_udt(column.as_udt().unwrap(), field_types) | ||
| } | ||
| _ => "ColumnType currently not implemented".into(), | ||
| }, | ||
| None => serde_json::Value::Null, | ||
| } | ||
| } | ||
|
|
||
| fn parse_udt( | ||
| udt: &[(String, Option<CqlValue>)], | ||
| field_types: &[(String, ColumnType)], | ||
| ) -> serde_json::Value { | ||
| let mut result = serde_json::Map::new(); | ||
|
|
||
| for (i, (field_name, field_value)) in udt.iter().enumerate() { | ||
| let field_type = &field_types[i].1; | ||
| let parsed_value = Self::parse_value(field_value, field_type); | ||
| result.insert(field_name.clone(), parsed_value); | ||
| } | ||
|
|
||
| serde_json::Value::Object(result) | ||
| } | ||
| } |
Oops, something went wrong.