rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-05-25 01:10:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: unify servers and mysql server in datanode (#172)

Browse Source 
* address PR comments

address PR comments

use 3306 for mysql server's default port

upgrade metric to version 0.20

move crate "servers" out of "common"

make mysql io threads count configurable in config file

add snafu backtrace for errors with source

use common-server error for mysql server

add test for grpc server

refactor testing codes

fix rustfmt check

start mysql server in datanode

move grpc server codes from datanode to common-servers

feat: unify servers

* rebase develop and resolve conflicts

* remove an unnecessary todo

Co-authored-by: luofucong <luofucong@greptime.com>
This commit is contained in:
LFC
2022-08-17 14:29:12 +08:00
committed by GitHub
parent 6d23118aa0
commit ccda17248e
53 changed files with 1036 additions and 768 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/common/runtime/Cargo.toml
View File

@@ -7,7 +7,7 @@ edition = "2021"
[dependencies]
common-error = { path = "../error" }
common-telemetry = { path = "../telemetry" }
metrics = "0.18"
metrics = "0.20"
once_cell = "1.12"
paste = "1.0"
snafu = { version = "0.7", features = ["backtraces"] }

27
src/common/servers/Cargo.toml
View File

@@ -1,27 +0,0 @@
[package]
name = "common-servers"
version = "0.1.0"
edition = "2021"
[dependencies]
async-trait = "0.1"
common-error = { path = "../error" }
common-recordbatch = { path = "../recordbatch" }
common-runtime = { path = "../runtime" }
common-telemetry = { path = "../telemetry" }
datatypes = { path = "../../datatypes"}
futures = "0.3"
metrics = "0.20"
num_cpus = "1.13"
opensrv-mysql = "0.1"
query = { path = "../../query" }
snafu = { version = "0.7", features = ["backtraces"] }
tokio = { version = "1.20", features = ["full"] }
tokio-stream = { version = "0.1", features = ["net"] }
[dev-dependencies]
common-base = { path = "../base" }
catalog = { path = "../../catalog" }
mysql_async = "0.30"
rand = "0.8"
test-util = { path = "../../../test-util" }

28
src/common/servers/src/error.rs
View File

@@ -1,28 +0,0 @@
use std::any::Any;
use common_error::prelude::*;
#[derive(Debug, Snafu)]
#[snafu(visibility(pub))]
pub enum Error {
#[snafu(display("MySQL server error, source: {}", source))]
MysqlServer { source: crate::mysql::error::Error },
}
pub type Result<T> = std::result::Result<T, Error>;
impl ErrorExt for Error {
fn status_code(&self) -> StatusCode {
match self {
Error::MysqlServer { .. } => StatusCode::Internal,
}
}
fn backtrace_opt(&self) -> Option<&Backtrace> {
ErrorCompat::backtrace(self)
}
fn as_any(&self) -> &dyn Any {
self
}
}

3
src/common/servers/src/lib.rs
View File

@@ -1,3 +0,0 @@
mod error;
pub mod mysql;
pub mod server;

60
src/common/servers/src/mysql/error.rs
View File

@@ -1,60 +0,0 @@
use std::any::Any;
use std::io;
use common_error::prelude::*;
#[derive(Debug, Snafu)]
#[snafu(visibility(pub))]
pub enum Error {
#[snafu(display("Internal error: {}", err_msg))]
Internal { err_msg: String },
#[snafu(display("Internal IO error, source: {}", source))]
InternalIo { source: io::Error },
#[snafu(display("Tokio IO error: {}, source: {}", err_msg, source))]
TokioIo { err_msg: String, source: io::Error },
#[snafu(display("Runtime resource error, source: {}", source))]
RuntimeResource {
source: common_runtime::error::Error,
},
#[snafu(display("Failed to convert vector, source: {}", source))]
VectorConversion { source: datatypes::error::Error },
#[snafu(display("Failed to collect recordbatch, source: {}", source))]
CollectRecordbatch {
source: common_recordbatch::error::Error,
},
}
pub type Result<T> = std::result::Result<T, Error>;
impl ErrorExt for Error {
fn status_code(&self) -> StatusCode {
match self {
Error::Internal { .. } | Error::InternalIo { .. } | Error::TokioIo { .. } => {
StatusCode::Unexpected
}
Error::VectorConversion { .. } | Error::CollectRecordbatch { .. } => {
StatusCode::Internal
}
Error::RuntimeResource { .. } => StatusCode::RuntimeResourcesExhausted,
}
}
fn backtrace_opt(&self) -> Option<&Backtrace> {
ErrorCompat::backtrace(self)
}
fn as_any(&self) -> &dyn Any {
self
}
}
impl From<io::Error> for Error {
fn from(e: io::Error) -> Self {
Error::InternalIo { source: e }
}
}

4
src/common/servers/src/mysql/mod.rs
View File

@@ -1,4 +0,0 @@
pub mod error;
pub mod mysql_instance;
pub mod mysql_server;
pub mod mysql_writer;

80
src/common/servers/src/mysql/mysql_instance.rs
View File

@@ -1,80 +0,0 @@
use std::io;
use std::sync::Arc;
use async_trait::async_trait;
use opensrv_mysql::AsyncMysqlShim;
use opensrv_mysql::ErrorKind;
use opensrv_mysql::ParamParser;
use opensrv_mysql::QueryResultWriter;
use opensrv_mysql::StatementMetaWriter;
use query::query_engine::Output;
use crate::mysql::error::{self, Result};
use crate::mysql::mysql_writer::MysqlResultWriter;
pub type MysqlInstanceRef = Arc<dyn MysqlInstance + Send + Sync>;
// TODO(LFC): Move to instance layer.
#[async_trait]
pub trait MysqlInstance {
async fn do_query(&self, query: &str) -> Result<Output>;
}
// An intermediate shim for executing MySQL queries.
pub struct MysqlInstanceShim {
mysql_instance: MysqlInstanceRef,
}
impl MysqlInstanceShim {
pub fn create(mysql_instance: MysqlInstanceRef) -> MysqlInstanceShim {
MysqlInstanceShim { mysql_instance }
}
}
#[async_trait]
impl<W: io::Write + Send + Sync> AsyncMysqlShim<W> for MysqlInstanceShim {
type Error = error::Error;
async fn on_prepare<'a>(
&'a mut self,
_: &'a str,
writer: StatementMetaWriter<'a, W>,
) -> Result<()> {
writer.error(
ErrorKind::ER_UNKNOWN_ERROR,
"prepare statement is not supported yet".as_bytes(),
)?;
Ok(())
}
async fn on_execute<'a>(
&'a mut self,
_: u32,
_: ParamParser<'a>,
writer: QueryResultWriter<'a, W>,
) -> Result<()> {
writer.error(
ErrorKind::ER_UNKNOWN_ERROR,
"prepare statement is not supported yet".as_bytes(),
)?;
Ok(())
}
async fn on_close<'a>(&'a mut self, _stmt_id: u32)
where
W: 'async_trait,
{
// do nothing because we haven't implemented prepare statement
}
async fn on_query<'a>(
&'a mut self,
query: &'a str,
writer: QueryResultWriter<'a, W>,
) -> Result<()> {
let output = self.mysql_instance.do_query(query).await;
let mut writer = MysqlResultWriter::new(writer);
writer.write(output).await
}
}

137
src/common/servers/src/mysql/mysql_server.rs
View File

@@ -1,137 +0,0 @@
use std::future::Future;
use std::net::SocketAddr;
use std::sync::Arc;
use async_trait::async_trait;
use common_runtime::Runtime;
use common_telemetry::logging::{error, info};
use futures::future::AbortHandle;
use futures::future::AbortRegistration;
use futures::future::Abortable;
use futures::StreamExt;
use opensrv_mysql::AsyncMysqlIntermediary;
use snafu::prelude::*;
use tokio;
use tokio::net::TcpStream;
use tokio::task::JoinHandle;
use tokio_stream::wrappers::TcpListenerStream;
use crate::error as server_error;
use crate::mysql::error::{self, Result};
use crate::mysql::mysql_instance::{MysqlInstanceRef, MysqlInstanceShim};
use crate::server::Server;
pub struct MysqlServer {
// `abort_handle` and `abort_registration` are used in pairs in shutting down MySQL server.
// They work like sender and receiver for aborting stream. When the server is shutting down,
// calling `abort_handle.abort()` will "notify" `abort_registration` to stop emitting new
// elements in the stream.
abort_handle: AbortHandle,
abort_registration: Option<AbortRegistration>,
// A handle holding the TCP accepting task.
join_handle: Option<JoinHandle<()>>,
mysql_handler: MysqlInstanceRef,
io_runtime: Arc<Runtime>,
}
impl MysqlServer {
/// Creates a new MySQL server with provided [MysqlInstance] and [Runtime].
pub fn create_server(
mysql_handler: MysqlInstanceRef,
io_runtime: Arc<Runtime>,
) -> Box<dyn Server> {
let (abort_handle, registration) = AbortHandle::new_pair();
Box::new(MysqlServer {
abort_handle,
abort_registration: Some(registration),
join_handle: None,
mysql_handler,
io_runtime,
})
}
async fn bind(addr: SocketAddr) -> Result<(TcpListenerStream, SocketAddr)> {
let listener = tokio::net::TcpListener::bind(addr)
.await
.context(error::TokioIoSnafu {
err_msg: format!("Failed to bind addr {}", addr),
})?;
// get actually bond addr in case input addr use port 0
let listener_addr = listener.local_addr()?;
Ok((TcpListenerStream::new(listener), listener_addr))
}
fn accept(&self, accepting_stream: Abortable<TcpListenerStream>) -> impl Future<Output = ()> {
let io_runtime = self.io_runtime.clone();
let mysql_handler = self.mysql_handler.clone();
accepting_stream.for_each(move |tcp_stream| {
let io_runtime = io_runtime.clone();
let mysql_handler = mysql_handler.clone();
async move {
match tcp_stream {
Err(error) => error!("Broken pipe: {}", error),
Ok(io_stream) => {
if let Err(error) = Self::handle(io_stream, io_runtime, mysql_handler) {
error!("Unexpected error when handling TcpStream: {:?}", error);
};
}
};
}
})
}
pub fn handle(
stream: TcpStream,
io_runtime: Arc<Runtime>,
mysql_handler: MysqlInstanceRef,
) -> Result<()> {
info!("MySQL connection coming from: {}", stream.peer_addr()?);
let shim = MysqlInstanceShim::create(mysql_handler);
// TODO(LFC): Relate "handler" with MySQL session; also deal with panics there.
let _handler = io_runtime.spawn(AsyncMysqlIntermediary::run_on(shim, stream));
Ok(())
}
}
#[async_trait]
impl Server for MysqlServer {
async fn shutdown(&mut self) -> server_error::Result<()> {
match self.join_handle.take() {
Some(join_handle) => {
self.abort_handle.abort();
if let Err(error) = join_handle.await {
error!("Unexpected error during shutdown MySQL server: {}", error);
} else {
info!("MySQL server is shutdown.")
}
Ok(())
}
None => error::InternalSnafu {
err_msg: "MySQL server is not started.",
}
.fail()
.context(server_error::MysqlServerSnafu),
}
}
async fn start(&mut self, listening: SocketAddr) -> server_error::Result<SocketAddr> {
match self.abort_registration.take() {
Some(registration) => {
let (stream, listener) = Self::bind(listening)
.await
.context(server_error::MysqlServerSnafu)?;
let stream = Abortable::new(stream, registration);
self.join_handle = Some(tokio::spawn(self.accept(stream)));
Ok(listener)
}
None => error::InternalSnafu {
err_msg: "MySQL server has been started.",
}
.fail()
.context(server_error::MysqlServerSnafu),
}
}
}

237
src/common/servers/src/mysql/mysql_writer.rs
View File

@@ -1,237 +0,0 @@
use std::io;
use common_recordbatch::{util, RecordBatch};
use datatypes::prelude::{ConcreteDataType, Value, VectorHelper};
use datatypes::schema::{ColumnSchema, SchemaRef};
use opensrv_mysql::{
Column, ColumnFlags, ColumnType, ErrorKind, OkResponse, QueryResultWriter, RowWriter,
};
use query::Output;
use snafu::prelude::*;
use crate::mysql::error::{self, Error, Result};
struct QueryResult {
recordbatches: Vec<RecordBatch>,
schema: SchemaRef,
}
pub struct MysqlResultWriter<'a, W: io::Write> {
// `QueryResultWriter` will be consumed when the write completed (see
// QueryResultWriter::completed), thus we use an option to wrap it.
inner: Option<QueryResultWriter<'a, W>>,
}
impl<'a, W: io::Write> MysqlResultWriter<'a, W> {
pub fn new(inner: QueryResultWriter<'a, W>) -> MysqlResultWriter<'a, W> {
MysqlResultWriter::<'a, W> { inner: Some(inner) }
}
pub async fn write(&mut self, output: Result<Output>) -> Result<()> {
let writer = self.inner.take().context(error::InternalSnafu {
err_msg: "inner MySQL writer is consumed",
})?;
match output {
Ok(output) => match output {
Output::RecordBatch(stream) => {
let schema = stream.schema().clone();
let recordbatches = util::collect(stream)
.await
.context(error::CollectRecordbatchSnafu)?;
let query_result = QueryResult {
recordbatches,
schema,
};
Self::write_query_result(query_result, writer)?
}
Output::AffectedRows(rows) => Self::write_affected_rows(writer, rows)?,
},
Err(error) => Self::write_query_error(error, writer)?,
}
Ok(())
}
fn write_affected_rows(writer: QueryResultWriter<W>, rows: usize) -> Result<()> {
writer.completed(OkResponse {
affected_rows: rows as u64,
..Default::default()
})?;
Ok(())
}
fn write_query_result(
query_result: QueryResult,
writer: QueryResultWriter<'a, W>,
) -> Result<()> {
if query_result.recordbatches.is_empty() {
writer.completed(OkResponse::default())?;
return Ok(());
}
match create_mysql_column_def(&query_result.schema) {
Ok(column_def) => {
let mut row_writer = writer.start(&column_def)?;
for recordbatch in &query_result.recordbatches {
Self::write_recordbatch(&mut row_writer, recordbatch)?;
}
row_writer.finish()?;
Ok(())
}
Err(error) => Self::write_query_error(error, writer),
}
}
fn write_recordbatch(row_writer: &mut RowWriter<W>, recordbatch: &RecordBatch) -> Result<()> {
let matrix = transpose(recordbatch)?;
for row in matrix.iter() {
for v in row.iter() {
match v {
Value::Null => row_writer.write_col(None::<u8>)?,
Value::Boolean(v) => row_writer.write_col(*v as i8)?,
Value::UInt8(v) => row_writer.write_col(v)?,
Value::UInt16(v) => row_writer.write_col(v)?,
Value::UInt32(v) => row_writer.write_col(v)?,
Value::UInt64(v) => row_writer.write_col(v)?,
Value::Int8(v) => row_writer.write_col(v)?,
Value::Int16(v) => row_writer.write_col(v)?,
Value::Int32(v) => row_writer.write_col(v)?,
Value::Int64(v) => row_writer.write_col(v)?,
Value::Float32(v) => row_writer.write_col(v.0)?,
Value::Float64(v) => row_writer.write_col(v.0)?,
Value::String(v) => row_writer.write_col(v.as_utf8())?,
Value::Binary(v) => row_writer.write_col(v.to_vec())?,
Value::Date(v) => row_writer.write_col(v)?,
Value::DateTime(v) => row_writer.write_col(v)?,
_ => {
return Err(Error::Internal {
err_msg: format!(
"cannot write value {:?} in mysql protocol: unimplemented",
v
),
})
}
}
}
row_writer.end_row()?;
}
Ok(())
}
fn write_query_error(error: Error, writer: QueryResultWriter<'a, W>) -> Result<()> {
writer.error(ErrorKind::ER_INTERNAL_ERROR, error.to_string().as_bytes())?;
Ok(())
}
}
fn create_mysql_column(column_schema: &ColumnSchema) -> Result<Column> {
let column_type = match column_schema.data_type {
ConcreteDataType::Null(_) => Ok(ColumnType::MYSQL_TYPE_NULL),
ConcreteDataType::Boolean(_) | ConcreteDataType::Int8(_) | ConcreteDataType::UInt8(_) => {
Ok(ColumnType::MYSQL_TYPE_TINY)
}
ConcreteDataType::Int16(_) | ConcreteDataType::UInt16(_) => {
Ok(ColumnType::MYSQL_TYPE_SHORT)
}
ConcreteDataType::Int32(_) | ConcreteDataType::UInt32(_) => Ok(ColumnType::MYSQL_TYPE_LONG),
ConcreteDataType::Int64(_) | ConcreteDataType::UInt64(_) => {
Ok(ColumnType::MYSQL_TYPE_LONGLONG)
}
ConcreteDataType::Float32(_) | ConcreteDataType::Float64(_) => {
Ok(ColumnType::MYSQL_TYPE_FLOAT)
}
ConcreteDataType::Binary(_) | ConcreteDataType::String(_) => {
Ok(ColumnType::MYSQL_TYPE_VARCHAR)
}
_ => error::InternalSnafu {
err_msg: format!(
"not implemented for column datatype {:?}",
column_schema.data_type
),
}
.fail(),
};
column_type.map(|column_type| Column {
column: column_schema.name.clone(),
coltype: column_type,
// TODO(LFC): Currently "table" and "colflags" are not relevant in MySQL server
// implementation, will revisit them again in the future.
table: "".to_string(),
colflags: ColumnFlags::empty(),
})
}
/// Creates MySQL columns definition from our column schema.
pub fn create_mysql_column_def(schema: &SchemaRef) -> Result<Vec<Column>> {
schema
.column_schemas()
.iter()
.map(create_mysql_column)
.collect()
}
/// RecordBatch organizes its values in columns while MySQL needs to write row by row.
/// This function creates a view of [Value]s organized in rows from RecordBatch (just like matrix
/// transpose, hence the function name), helping us write RecordBatch to MySQL.
fn transpose(recordbatch: &RecordBatch) -> Result<Vec<Vec<Value>>> {
let recordbatch = &recordbatch.df_recordbatch;
let rows = recordbatch.num_rows();
let columns = recordbatch.num_columns();
let mut matrix = vec![vec![Value::Null; columns]; rows];
for column in 0..columns {
let array = recordbatch.column(column);
let vector = VectorHelper::try_into_vector(array).context(error::VectorConversionSnafu)?;
// Clippy suggests us to use "matrix.iter_mut().enumerate().take(rows)", which is not wanted.
#[allow(clippy::needless_range_loop)]
for row in 0..rows {
matrix[row][column] = vector.get(row);
}
}
Ok(matrix)
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use common_base::bytes::StringBytes;
use datatypes::prelude::*;
use datatypes::schema::Schema;
use datatypes::vectors::{StringVector, UInt32Vector};
use super::*;
#[test]
fn test_transpose() {
let column_schemas = vec![
ColumnSchema::new("numbers", ConcreteDataType::uint32_datatype(), false),
ColumnSchema::new("strings", ConcreteDataType::string_datatype(), true),
];
let schema = Arc::new(Schema::new(column_schemas));
let columns: Vec<VectorRef> = vec![
Arc::new(UInt32Vector::from_slice(vec![1, 2, 3, 4])),
Arc::new(StringVector::from(vec![
None,
Some("hello"),
Some("greptime"),
None,
])),
];
let recordbatch = RecordBatch::new(schema, columns).unwrap();
let matrix = transpose(&recordbatch).unwrap();
assert_eq!(4, matrix.len());
assert_eq!(vec![Value::UInt32(1), Value::Null], matrix[0]);
assert_eq!(
vec![Value::UInt32(2), Value::String(StringBytes::from("hello"))],
matrix[1]
);
assert_eq!(
vec![
Value::UInt32(3),
Value::String(StringBytes::from("greptime"))
],
matrix[2]
);
assert_eq!(vec![Value::UInt32(4), Value::Null], matrix[3]);
}
}

11
src/common/servers/src/server.rs
View File

@@ -1,11 +0,0 @@
use std::net::SocketAddr;
use async_trait::async_trait;
use crate::error::Result;
#[async_trait]
pub trait Server: Send {
async fn shutdown(&mut self) -> Result<()>;
async fn start(&mut self, listening: SocketAddr) -> Result<SocketAddr>;
}

1
src/common/servers/tests/mod.rs
View File

@@ -1 +0,0 @@
mod mysql;

269
src/common/servers/tests/mysql/mod.rs
View File

@@ -1,269 +0,0 @@
use std::sync::Arc;
use datatypes::prelude::*;
use datatypes::schema::ColumnSchema;
use datatypes::vectors::{
BinaryVector, BooleanVector, Float32Vector, Float64Vector, Int16Vector, Int32Vector,
Int64Vector, Int8Vector, NullVector, StringVector, UInt16Vector, UInt32Vector, UInt64Vector,
UInt8Vector,
};
use mysql_async::prelude::FromRow;
use mysql_async::FromRowError;
use mysql_async::Value as MysqlValue;
use opensrv_mysql::ColumnType;
mod mysql_server_test;
mod mysql_writer_test;
pub struct TestingData {
column_schemas: Vec<ColumnSchema>,
mysql_columns_def: Vec<ColumnType>,
columns: Vec<VectorRef>,
mysql_text_output_rows: Vec<Vec<Value>>,
}
impl TestingData {
fn new(
column_schemas: Vec<ColumnSchema>,
mysql_columns_def: Vec<ColumnType>,
columns: Vec<VectorRef>,
mysql_text_output_rows: Vec<Vec<Value>>,
) -> Self {
// Check input columns have same size,
assert_eq!(column_schemas.len(), mysql_columns_def.len());
assert_eq!(column_schemas.len(), columns.len());
// and all columns length are equal
assert!(columns.windows(2).all(|x| x[0].len() == x[1].len()));
// and all output rows width are equal
assert!(mysql_text_output_rows
.windows(2)
.all(|x| x[0].len() == x[1].len()));
// and the rows' columns size equals to input columns size.
assert_eq!(columns.first().unwrap().len(), mysql_text_output_rows.len());
TestingData {
column_schemas,
mysql_columns_def,
columns,
mysql_text_output_rows,
}
}
}
#[derive(Debug)]
struct MysqlTextRow {
values: Vec<Value>,
}
impl FromRow for MysqlTextRow {
fn from_row_opt(row: mysql_async::Row) -> Result<Self, FromRowError>
where
Self: Sized,
{
let mut values = Vec::with_capacity(row.len());
for i in 0..row.len() {
let value = if let Some(mysql_value) = row.as_ref(i) {
match mysql_value {
MysqlValue::NULL => Value::Null,
MysqlValue::Bytes(v) => Value::from(v.to_vec()),
_ => unreachable!(),
}
} else {
Value::Null
};
values.push(value);
}
Ok(MysqlTextRow { values })
}
}
pub fn all_datatype_testing_data() -> TestingData {
let column_schemas = vec![
ColumnSchema::new("nulls", ConcreteDataType::null_datatype(), true),
ColumnSchema::new("bools", ConcreteDataType::boolean_datatype(), true),
ColumnSchema::new("int8s", ConcreteDataType::int8_datatype(), true),
ColumnSchema::new("int16s", ConcreteDataType::int16_datatype(), true),
ColumnSchema::new("int32s", ConcreteDataType::int32_datatype(), true),
ColumnSchema::new("int64s", ConcreteDataType::int64_datatype(), true),
ColumnSchema::new("uint8s", ConcreteDataType::uint8_datatype(), true),
ColumnSchema::new("uint16s", ConcreteDataType::uint16_datatype(), true),
ColumnSchema::new("uint32s", ConcreteDataType::uint32_datatype(), true),
ColumnSchema::new("uint64s", ConcreteDataType::uint64_datatype(), true),
ColumnSchema::new("float32s", ConcreteDataType::float32_datatype(), true),
ColumnSchema::new("float64s", ConcreteDataType::float64_datatype(), true),
ColumnSchema::new("binaries", ConcreteDataType::binary_datatype(), true),
ColumnSchema::new("strings", ConcreteDataType::string_datatype(), true),
];
let mysql_columns_def = vec![
ColumnType::MYSQL_TYPE_NULL,
ColumnType::MYSQL_TYPE_TINY,
ColumnType::MYSQL_TYPE_TINY,
ColumnType::MYSQL_TYPE_SHORT,
ColumnType::MYSQL_TYPE_LONG,
ColumnType::MYSQL_TYPE_LONGLONG,
ColumnType::MYSQL_TYPE_TINY,
ColumnType::MYSQL_TYPE_SHORT,
ColumnType::MYSQL_TYPE_LONG,
ColumnType::MYSQL_TYPE_LONGLONG,
ColumnType::MYSQL_TYPE_FLOAT,
ColumnType::MYSQL_TYPE_FLOAT,
ColumnType::MYSQL_TYPE_VARCHAR,
ColumnType::MYSQL_TYPE_VARCHAR,
];
let columns: Vec<VectorRef> = vec![
Arc::new(NullVector::new(4)),
Arc::new(BooleanVector::from(vec![
Some(true),
None,
Some(false),
None,
])),
Arc::new(Int8Vector::from(vec![
Some(i8::MIN),
None,
Some(i8::MAX),
None,
])),
Arc::new(Int16Vector::from(vec![
Some(i16::MIN),
None,
Some(i16::MAX),
None,
])),
Arc::new(Int32Vector::from(vec![
Some(i32::MIN),
None,
Some(i32::MAX),
None,
])),
Arc::new(Int64Vector::from(vec![
Some(i64::MIN),
None,
Some(i64::MAX),
None,
])),
Arc::new(UInt8Vector::from(vec![
Some(u8::MIN),
None,
Some(u8::MAX),
None,
])),
Arc::new(UInt16Vector::from(vec![
Some(u16::MIN),
None,
Some(u16::MAX),
None,
])),
Arc::new(UInt32Vector::from(vec![
Some(u32::MIN),
None,
Some(u32::MAX),
None,
])),
Arc::new(UInt64Vector::from(vec![
Some(u64::MIN),
None,
Some(u64::MAX),
None,
])),
Arc::new(Float32Vector::from(vec![
Some(-1.123456_f32),
None,
Some(1.654321),
None,
])),
Arc::new(Float64Vector::from(vec![
Some(-10.123456_f64),
None,
Some(10.654321),
None,
])),
Arc::new(BinaryVector::from(vec![
None,
Some("hello".as_bytes().to_vec()),
Some("greptime".as_bytes().to_vec()),
None,
])),
Arc::new(StringVector::from(vec![
Some("hola"),
None,
None,
Some("GT"),
])),
];
// Because we can only use MySQL text protocol (binary protocol requires prepared statement,
// which we are not implemented yet), every MysqlValue is of type "Bytes"
let mysql_text_output_rows = vec![
vec![
Value::Null,
Value::from("1".as_bytes()),
Value::from(i8::MIN.to_string().as_bytes()),
Value::from(i16::MIN.to_string().as_bytes()),
Value::from(i32::MIN.to_string().as_bytes()),
Value::from(i64::MIN.to_string().as_bytes()),
Value::from(u8::MIN.to_string().as_bytes()),
Value::from(u16::MIN.to_string().as_bytes()),
Value::from(u32::MIN.to_string().as_bytes()),
Value::from(u64::MIN.to_string().as_bytes()),
Value::from((-1.123456_f32).to_string().as_bytes()),
Value::from((-10.123456_f64).to_string().as_bytes()),
Value::Null,
Value::from("hola".as_bytes()),
],
vec![
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::from("hello".as_bytes()),
Value::Null,
],
vec![
Value::Null,
Value::from("0".as_bytes()),
Value::from(i8::MAX.to_string().as_bytes()),
Value::from(i16::MAX.to_string().as_bytes()),
Value::from(i32::MAX.to_string().as_bytes()),
Value::from(i64::MAX.to_string().as_bytes()),
Value::from(u8::MAX.to_string().as_bytes()),
Value::from(u16::MAX.to_string().as_bytes()),
Value::from(u32::MAX.to_string().as_bytes()),
Value::from(u64::MAX.to_string().as_bytes()),
Value::from(1.654321_f32.to_string().as_bytes()),
Value::from(10.654321_f64.to_string().as_bytes()),
Value::from("greptime".as_bytes()),
Value::Null,
],
vec![
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::Null,
Value::from("GT".as_bytes()),
],
];
TestingData::new(
column_schemas,
mysql_columns_def,
columns,
mysql_text_output_rows,
)
}

223
src/common/servers/tests/mysql/mysql_server_test.rs
View File

@@ -1,223 +0,0 @@
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use catalog::memory::{MemoryCatalogList, MemoryCatalogProvider, MemorySchemaProvider};
use catalog::{
CatalogList, CatalogProvider, SchemaProvider, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME,
};
use common_recordbatch::RecordBatch;
use common_runtime::Builder as RuntimeBuilder;
use common_servers::mysql::error::{Result, RuntimeResourceSnafu};
use common_servers::mysql::mysql_instance::MysqlInstance;
use common_servers::mysql::mysql_server::MysqlServer;
use common_servers::server::Server;
use datatypes::schema::Schema;
use mysql_async::prelude::*;
use query::{Output, QueryEngineFactory, QueryEngineRef};
use rand::rngs::StdRng;
use rand::Rng;
use snafu::prelude::*;
use test_util::MemTable;
use crate::mysql::{all_datatype_testing_data, MysqlTextRow, TestingData};
fn create_mysql_server(table: MemTable) -> Result<Box<dyn Server>> {
let table_name = table.table_name().to_string();
let table = Arc::new(table);
let schema_provider = Arc::new(MemorySchemaProvider::new());
schema_provider.register_table(table_name, table).unwrap();
let catalog_provider = Arc::new(MemoryCatalogProvider::new());
catalog_provider.register_schema(DEFAULT_SCHEMA_NAME.to_string(), schema_provider);
let catalog_list = Arc::new(MemoryCatalogList::default());
catalog_list.register_catalog(DEFAULT_CATALOG_NAME.to_string(), catalog_provider);
let factory = QueryEngineFactory::new(catalog_list);
let query_engine = factory.query_engine().clone();
let mysql_instance = Arc::new(DummyMysqlInstance { query_engine });
let io_runtime = Arc::new(
RuntimeBuilder::default()
.worker_threads(4)
.thread_name("mysql-io-handlers")
.build()
.context(RuntimeResourceSnafu)?,
);
Ok(MysqlServer::create_server(mysql_instance, io_runtime))
}
#[tokio::test]
async fn test_start_mysql_server() -> Result<()> {
let table = MemTable::default_numbers_table();
let mut mysql_server = create_mysql_server(table)?;
let listening = "127.0.0.1:0".parse::<SocketAddr>().unwrap();
let result = mysql_server.start(listening).await;
assert!(result.is_ok());
let result = mysql_server.start(listening).await;
assert!(result
.unwrap_err()
.to_string()
.contains("MySQL server has been started."));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_shutdown_mysql_server() -> Result<()> {
common_telemetry::init_default_ut_logging();
let table = MemTable::default_numbers_table();
let mut mysql_server = create_mysql_server(table)?;
let result = mysql_server.shutdown().await;
assert!(result
.unwrap_err()
.to_string()
.contains("MySQL server is not started."));
let listening = "127.0.0.1:0".parse::<SocketAddr>().unwrap();
let server_addr = mysql_server.start(listening).await.unwrap();
let server_port = server_addr.port();
let mut join_handles = vec![];
for _ in 0..2 {
join_handles.push(tokio::spawn(async move {
for _ in 0..1000 {
match create_connection(server_port).await {
Ok(mut connection) => {
let result: u32 = connection
.query_first("SELECT uint32s FROM numbers LIMIT 1")
.await
.unwrap()
.unwrap();
assert_eq!(result, 0);
tokio::time::sleep(Duration::from_millis(10)).await;
}
Err(e) => return Err(e),
}
}
Ok(())
}))
}
tokio::time::sleep(Duration::from_millis(100)).await;
let result = mysql_server.shutdown().await;
assert!(result.is_ok());
for handle in join_handles.iter_mut() {
let result = handle.await.unwrap();
assert!(result.is_err());
let error = result.unwrap_err().to_string();
assert!(error.contains("Connection refused") || error.contains("Connection reset by peer"));
}
Ok(())
}
#[tokio::test]
async fn test_query_all_datatypes() -> Result<()> {
common_telemetry::init_default_ut_logging();
let TestingData {
column_schemas,
mysql_columns_def,
columns,
mysql_text_output_rows,
} = all_datatype_testing_data();
let schema = Arc::new(Schema::new(column_schemas.clone()));
let recordbatch = RecordBatch::new(schema, columns).unwrap();
let table = MemTable::new("all_datatypes", recordbatch);
let mut mysql_server = create_mysql_server(table)?;
let listening = "127.0.0.1:0".parse::<SocketAddr>().unwrap();
let server_addr = mysql_server.start(listening).await.unwrap();
let mut connection = create_connection(server_addr.port()).await.unwrap();
let mut result = connection
.query_iter("SELECT * FROM all_datatypes LIMIT 3")
.await
.unwrap();
let columns = result.columns().unwrap();
assert_eq!(column_schemas.len(), columns.len());
for (i, column) in columns.iter().enumerate() {
assert_eq!(mysql_columns_def[i], column.column_type());
assert_eq!(column_schemas[i].name, column.name_str());
}
let rows = result.collect::<MysqlTextRow>().await.unwrap();
assert_eq!(3, rows.len());
for (expected, actual) in mysql_text_output_rows.iter().take(3).zip(rows.iter()) {
assert_eq!(expected, &actual.values);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_query_concurrently() -> Result<()> {
common_telemetry::init_default_ut_logging();
let table = MemTable::default_numbers_table();
let mut mysql_server = create_mysql_server(table)?;
let listening = "127.0.0.1:0".parse::<SocketAddr>().unwrap();
let server_addr = mysql_server.start(listening).await.unwrap();
let server_port = server_addr.port();
let threads = 4;
let expect_executed_queries_per_worker = 1000;
let mut join_handles = vec![];
for _ in 0..threads {
join_handles.push(tokio::spawn(async move {
let mut rand: StdRng = rand::SeedableRng::from_entropy();
let mut connection = create_connection(server_port).await.unwrap();
for _ in 0..expect_executed_queries_per_worker {
let expected: u32 = rand.gen_range(0..100);
let result: u32 = connection
.query_first(format!(
"SELECT uint32s FROM numbers WHERE uint32s = {}",
expected
))
.await
.unwrap()
.unwrap();
assert_eq!(result, expected);
let should_recreate_conn = expected == 1;
if should_recreate_conn {
connection = create_connection(server_port).await.unwrap();
}
}
expect_executed_queries_per_worker
}))
}
let mut total_pending_queries = threads * expect_executed_queries_per_worker;
for handle in join_handles.iter_mut() {
total_pending_queries -= handle.await.unwrap();
}
assert_eq!(0, total_pending_queries);
Ok(())
}
async fn create_connection(port: u16) -> mysql_async::Result<mysql_async::Conn> {
let opts = mysql_async::OptsBuilder::default()
.ip_or_hostname("127.0.0.1")
.tcp_port(port)
.prefer_socket(false)
.wait_timeout(Some(1000));
mysql_async::Conn::new(opts).await
}
struct DummyMysqlInstance {
query_engine: QueryEngineRef,
}
#[async_trait]
impl MysqlInstance for DummyMysqlInstance {
async fn do_query(&self, query: &str) -> Result<Output> {
let plan = self.query_engine.sql_to_plan(query).unwrap();
Ok(self.query_engine.execute(&plan).await.unwrap())
}
}

34
src/common/servers/tests/mysql/mysql_writer_test.rs
View File

@@ -1,34 +0,0 @@
use std::sync::Arc;
use common_servers::mysql::mysql_writer::create_mysql_column_def;
use datatypes::prelude::*;
use datatypes::schema::{ColumnSchema, Schema};
use crate::mysql::{all_datatype_testing_data, TestingData};
#[test]
fn test_create_mysql_column_def() {
let TestingData {
column_schemas,
mysql_columns_def,
..
} = all_datatype_testing_data();
let schema = Arc::new(Schema::new(column_schemas.clone()));
let columns_def = create_mysql_column_def(&schema).unwrap();
assert_eq!(column_schemas.len(), columns_def.len());
for (i, column_def) in columns_def.iter().enumerate() {
let column_schema = &column_schemas[i];
assert_eq!(column_schema.name, column_def.column);
let expected_coltype = mysql_columns_def[i];
assert_eq!(column_def.coltype, expected_coltype);
}
let column_schemas = vec![ColumnSchema::new(
"lists",
ConcreteDataType::list_datatype(ConcreteDataType::string_datatype()),
true,
)];
let schema = Arc::new(Schema::new(column_schemas));
assert!(create_mysql_column_def(&schema).is_err());
}

4
src/common/telemetry/Cargo.toml
View File

@@ -13,8 +13,8 @@ deadlock_detection=["parking_lot"]
backtrace = "0.3"
common-error = { path = "../error" }
console-subscriber = { version = "0.1", optional = true }
metrics = "0.18"
metrics-exporter-prometheus = { version = "0.9", default-features = false }
metrics = "0.20"
metrics-exporter-prometheus = { version = "0.11", default-features = false }
once_cell = "1.10"
opentelemetry = { version = "0.17", default-features = false, features = ["trace", "rt-tokio"] }
opentelemetry-jaeger = { version = "0.16", features = ["rt-tokio"] }

11
src/common/telemetry/src/metric.rs
View File

@@ -21,7 +21,16 @@ fn init_prometheus_recorder() {
let recorder = PrometheusBuilder::new().build_recorder();
let mut h = PROMETHEUS_HANDLE.as_ref().write().unwrap();
*h = Some(recorder.handle());
metrics::clear_recorder();
// TODO(LFC): separate metrics for testing and metrics for production
// `clear_recorder` is likely not expected to be called in production code, recorder should be
// globally unique and used throughout the whole lifetime of an application.
// It's marked as "unsafe" since [this PR](https://github.com/metrics-rs/metrics/pull/302), and
// "metrics" version also upgraded to 0.19.
// A quick look in the metrics codes suggests that the "unsafe" call is of no harm. However,
// it required a further investigation in how to use metric properly.
unsafe {
metrics::clear_recorder();
}
match metrics::set_boxed_recorder(Box::new(recorder)) {
Ok(_) => (),
Err(err) => crate::warn!("Install prometheus recorder failed, cause: {}", err),