jonhoo / msql-srv Goto Github PK

Bindings for writing a server that can act as MySQL/MariaDB

License: Apache License 2.0

Rust 100.00%

msql-srv's Introduction

msql-srv-rs

Bindings for emulating a MySQL/MariaDB server.

When developing new databases or caching layers, it can be immensely useful to test your system using existing applications. However, this often requires significant work modifying applications to use your database over the existing ones. This crate solves that problem by acting as a MySQL server, and delegating operations such as querying and query execution to user-defined logic.

To start, implement MysqlShim for your backend, and create a MysqlIntermediary over an instance of your backend and a connection stream. The appropriate methods will be called on your backend whenever a client issues a QUERY, PREPARE, or EXECUTE command, and you will have a chance to respond appropriately. For example, to write a shim that always responds to all commands with a "no results" reply:

extern crate mysql;
use msql_srv::*;
use mysql::prelude::*;

struct Backend;
impl<W: io::Write> MysqlShim<W> for Backend {
    type Error = io::Error;

    fn on_prepare(&mut self, _: &str, info: StatementMetaWriter<W>) -> io::Result<()> {
        info.reply(42, &[], &[])
    }
    fn on_execute(
        &mut self,
        _: u32,
        _: ParamParser,
        results: QueryResultWriter<W>,
    ) -> io::Result<()> {
        results.completed(0, 0)
    }
    fn on_close(&mut self, _: u32) {}

    fn on_init(&mut self, _: &str, writer: InitWriter<W>) -> io::Result<()> { Ok(()) }

    fn on_query(&mut self, _: &str, results: QueryResultWriter<W>) -> io::Result<()> {
        let cols = [
            Column {
                table: "foo".to_string(),
                column: "a".to_string(),
                coltype: ColumnType::MYSQL_TYPE_LONGLONG,
                colflags: ColumnFlags::empty(),
            },
            Column {
                table: "foo".to_string(),
                column: "b".to_string(),
                coltype: ColumnType::MYSQL_TYPE_STRING,
                colflags: ColumnFlags::empty(),
            },
        ];

        let mut rw = results.start(&cols)?;
        rw.write_col(42)?;
        rw.write_col("b's value")?;
        rw.finish()
    }
}

fn main() {
    let listener = net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();

    let jh = thread::spawn(move || {
        if let Ok((s, _)) = listener.accept() {
            MysqlIntermediary::run_on_tcp(Backend, s).unwrap();
        }
    });

    let mut db = mysql::Conn::new(&format!("mysql://127.0.0.1:{}", port)).unwrap();
    assert_eq!(db.ping(), true);
    assert_eq!(db.query_iter("SELECT a, b FROM foo").unwrap().count(), 1);
    drop(db);
    jh.join().unwrap();
}

msql-srv's People

Contributors

Stargazers

Watchers

msql-srv's Issues

Return from `MysqlShim` methods instead of passing writers?

Currently MysqlShim methods on_prepare, on_execute and on_query return Result<(), Self::Error> and use a writer value to write data.

This requires the methods and the trait itself to depend on the generic W: io::Write parameter. However none of the writer method signatures depend on W. So it should be possible to perform the operations that depend on W outside of the trait's implementations.

The first thing that comes to mind is returning values from the methods instead of passing them to writers. This would work well for on_prepare and on_execute since the writers are supposed to be only used once there as far as I can tell.

on_query is harder since it feeds values one by one. I imagine this must work for big queries where it doesn't make sense to collect a single value in memory before returning.

One solution is to return an async stream, but that makes the rest of the crate depend on an async runtime and probably means replacing all of the std::io stuff.

Another is to provide a non-generic channel-like writer that merely accepts values. Not sure if there are any caveats there.

About setup steps

Hi,
Since I'm new to rust, is it okay if somebody could help me with how to run those examples after cargo build the rlib file? I would really appreciate it! Thanks!

error: failed to select a version for `time`

When adding the msql-srv dependency to my project, I get this error

error: failed to select a version for `time`.
    ... required by package `cookie v0.14.1`
    ... which is depended on by `actix-http v2.0.0`
    ... which is depended on by `actix-web v3.0.0`
    ... which is depended on by `qz-server v0.1.0`
versions that meet the requirements `^0.2.11` are: 0.2.24, 0.2.25, 0.2.23, 0.2.22, 0.2.21, 0.2.20, 0.2.19, 0.2.18, 0.2.17, 0.2.16, 0.2.15, 0.2.14, 0.2.13, 0.2.12, 0.2.11

all possible versions conflict with previously selected packages.

  previously selected package `time v0.2.7`
    ... which is depended on by `msql-srv v0.9.2`
    ... which is depended on by `qz-drivers v0.1.0`
    ... which is depended on by `qz-server v0.1.0`

Wondering if anyone else is seeing this?

can not connect with JDBC driver

I am trying to run oltpbench on a SQL server written with msql-srv. But it does not connect. It throws error at this code block
https://github.com/jonhoo/msql-srv/blob/master/src/lib.rs#L255-L259

oltpbench uses JDBC drivers. Do you know what is going on?

mysql terminal session or rust crate works just fine.

Any documentation for the architecture?

Hello,

I'm trying to implement something similar for PostgreSQL but am unsure of how to go about doing it. Is there any documentation for the architecture? What patterns did you use, any other projects you use for inspiration?

Support multiple resultsets

If a received query produces multiple result sets (e.g., because it contains multiple ;-separated queries), the MySQL binary protocol uses the MORE_RESULTS_EXISTS flag. Specifically, it sets the SERVER_MORE_RESULTS_EXISTS flag on the OK packet that follows each non-final result set. msql-srv does not yet support this feature.

The way to add support for this would likely be to add a continue method on RowWriter which finished up the current resultset with MORE_RESULTS_EXISTS set, and then returns another QueryResultWriter. There would also need to be something equivalent available for QueryResultWriter::completed and QueryResultWriter::error.

Incorrect handling of server system variables

A recent update of mysql_async (0.32.2) consolidated multiple server system variable queries into one:

let settings: Option<Row> = if read_socket || read_max_allowed_packet || read_wait_timeout {
    self.query_internal("SELECT @@socket, @@max_allowed_packet, @@wait_timeout")
        .await?
} else {
    None
};

https://github.com/blackbeam/mysql_async/blob/e6bbf7c776374d0067c0164245e9158b5c75f7e7/src/conn/mod.rs#L970

This query is not handled correctly, as the responsible for this accepts only the following query:

SELECT @@max_allowed_packet

msql-srv/src/lib.rs

Line 419 in 9d85e6b

if q.starts_with(b"SELECT @@") || q.starts_with(b"select @@") {

This in turn triggers a bug in mysql_async, preventing the creation of a new connection (which is fixed in the next version blackbeam/mysql_async#263). However, the handling of the max_allowed_packet in msql-srv should probably be corrected or removed.

Assert failure

It turns out that the assert here https://github.com/jonhoo/msql-srv/blob/master/src/packet.rs#L80 can happen after all.

I can't reliably reproduce this yet, but have seen it during happen during various testing. I will investigate.

mysql connector 5.1.48 throws exception on getConnection()

It only happens when I am stressing the server. This loop easily reproduces the exception on my Linux machine.

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

class Sql {
    public static void main(String[] args) throws SQLException {
        for(int i = 0; i < 1000; ++i) {
            Connection conn = DriverManager.getConnection(
                "jdbc:mysql://localhost:3306",
                "",
                ""
            );
        }
    }
}

Replace on_query() method in serve_one.rs to accept JDBC connections.

fn on_query(&mut self, sql: &str, results: QueryResultWriter<W>) -> io::Result<()> {
        println!("sqlquery {}", sql);
        use sqlparser::ast;
        use sqlparser::ast::Statement::Query;
        use sqlparser::dialect;
        use sqlparser::parser::Parser;
        let parsed_sql = Parser::parse_sql(&dialect::AnsiDialect {}, &sql)
            .or_else(|_| Parser::parse_sql(&dialect::GenericDialect {}, &sql))
            .or_else(|_| Parser::parse_sql(&dialect::MsSqlDialect {}, &sql))
            .or_else(|_| Parser::parse_sql(&dialect::MySqlDialect {}, &sql))
            .or_else(|_| Parser::parse_sql(&dialect::PostgreSqlDialect {}, &sql))
            .unwrap();
        match &parsed_sql[0] {
            Query(query) => {
                match &query.body {
                    ast::SetExpr::Select(ref s) => {
                        let mut col_writer = Vec::new();

                        for proj in &s.projection {
                            if let ast::SelectItem::ExprWithAlias { alias, .. } = proj {
                                let col = msql_srv::Column {
                                    table: "".into(),
                                    column: alias.value.clone(),
                                    coltype: msql_srv::ColumnType::MYSQL_TYPE_VAR_STRING,
                                    colflags: msql_srv::ColumnFlags::empty(),
                                };
                                col_writer.push(col);
                            }
                        }

                        let mut rw = results.start(&col_writer).unwrap();

                        for proj in &s.projection {
                            if let ast::SelectItem::ExprWithAlias { alias, .. } = proj {
                                // TODO : also handle STOCK_COUNT
                                let val = match alias.value.as_str() {
                                        "auto_increment_increment" => "1",
                                        "character_set_client" => "utf8",
                                        "character_set_connection" => "utf8",
                                        "character_set_results" => "utf8",
                                        "character_set_server" => "utf8mb4",
                                        "collation_server" => "utf8mb4_0900_ai_ci",
                                        "collation_connection" => "utf8_general_ci",
                                        "init_connect" => "",
                                        "interactive_timeout" => "28800",
                                        "license" => "GPL",
                                        "lower_case_table_names" => "0",
                                        "max_allowed_packet" => "67108864",
                                        "net_buffer_length" => "16384",
                                        "net_write_timeout" => "60",
                                        "sql_mode" => "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION",
                                        "system_time_zone" => "UTC",
                                        "time_zone" => "SYSTEM",
                                        "wait_timeout" => "28800",
                                        "language" => "eng",
                                        "query_cache_size" => "",
                                        "query_cache_type" => "",
                                        "transaction_isolation" => "SERIALIZATION",
                                        e => unreachable!("NEW COLUMN {}", e),
                                    };
                                // println!("{} : {}", proj, val);
                                rw.write_col(val).unwrap();
                            }
                        }

                        rw.end_row().unwrap();

                        rw.finish()
                    }
                    _ => unimplemented!(),
                }
            }
            _ => results.completed(1, 0),
        }
    }

Now run them.

$ cargo run --release --example server_one &
$ javac Sql.java
$ java -cp .:./mysql-connector-java-5.1.48/mysql-connector-java-5.1.48.jar Sql
Exception in thread "main" java.sql.SQLException: ResultSet is from UPDATE. No Data.
	at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:965)
	at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:898)
	at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:887)
	at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:861)
	at com.mysql.jdbc.ResultSetImpl.next(ResultSetImpl.java:6292)
	at com.mysql.jdbc.ConnectionImpl.loadServerVariables(ConnectionImpl.java:3768)
	at com.mysql.jdbc.ConnectionImpl.initializePropsFromServer(ConnectionImpl.java:3196)
	at com.mysql.jdbc.ConnectionImpl.connectOneTryOnly(ConnectionImpl.java:2233)
	at com.mysql.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:2015)
	at com.mysql.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:768)
	at com.mysql.jdbc.JDBC4Connection.<init>(JDBC4Connection.java:47)
	at sun.reflect.GeneratedConstructorAccessor3.newInstance(Unknown Source)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at com.mysql.jdbc.Util.handleNewInstance(Util.java:425)
	at com.mysql.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:385)
	at com.mysql.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:323)
	at java.sql.DriverManager.getConnection(DriverManager.java:664)
	at java.sql.DriverManager.getConnection(DriverManager.java:247)
	at Sql.main(Sql.java:8)

PS. I know there are other issues with JDBC driver already. I am trying to collect the bugs to fix them later. 5.1.48 is the last version that does not complain about CLIENT_SECURE_CONNECTION (mit-pdos/noria-mysql#15).

Multiple rows throws error from select query

When a select returns multiple rows, it throws "row has fewer columns than specification" !

I am following your postgresql example.

TLS support

Is there any interest in supporting SSL encrypted mysql connections?

I have added this locally for my own purposes, but it's a bit ugly at the moment.

If it's wanted, I'm happy to tidy it up and submit a PR.

let shims know about USE schema commands

It would be useful for the shims to be notified of the client issuing a 0x02 COM_INIT_DB / Command::Init(&'a [u8]). This command is sent by users on the mysql client to switch the connection context to another schema.

Currently this command is handled in lib.rs by sending the OK packet without notifying the shim.

A potential solution is to add on_init_db(&mut self, schema: &str) to the MysqlShim trait.

This would impact users of this package.

Would you consider accepting a contribution for this?

Can't Parse the database info From Handshake.

In the Commands.rs, when the handshake is HandshakeResponse41,I want to parse the database when the client log in like mysql -h localhost -D test -utest -pxxx,but the packet contains nothing about database.

Remove support for old handshake

Currently, the library includes the following code to handle an old version of the client handshake

msql-srv/src/commands.rs

Lines 47 to 63 in 26a5c35

 } else { 

 // HandshakeResponse320 

 let (i, maxps1) = nom::number::complete::le_u16(i)?; 

 let (i, maxps2) = nom::number::complete::le_u8(i)?; 

 let maxps = (maxps2 as u32) << 16 | maxps1 as u32; 

 let (i, username) = nom::bytes::complete::take_until(&b"\0"[..])(i)?; 

 Ok(( 

 i, 

 ClientHandshake { 

 capabilities: CapabilityFlags::from_bits_truncate(cap as u32), 

 maxps, 

 collation: 0, 

 username: Some(username), 

 }, 

 )) 

 }

This gives the impression that the library supports that version of the protocol, but that is probably not the case. Since the old handshake was replaced in version 3.21.0, released in 1998, I think it is safe to assume no MySQL client out there will be using it. IMO it would make sense to remove it from the codebase, both to simplify the code and to avoid confusion for people less familiar with the MySQL protocol.

	} else {
	// HandshakeResponse320
	let (i, maxps1) = nom::number::complete::le_u16(i)?;
	let (i, maxps2) = nom::number::complete::le_u8(i)?;
	let maxps = (maxps2 as u32) << 16 \| maxps1 as u32;
	let (i, username) = nom::bytes::complete::take_until(&b"\0"[..])(i)?;

	Ok((
	i,
	ClientHandshake {
	capabilities: CapabilityFlags::from_bits_truncate(cap as u32),
	maxps,
	collation: 0,
	username: Some(username),
	},
	))
	}

jonhoo / msql-srv Goto Github PK

msql-srv's Introduction

msql-srv-rs

msql-srv's People

Contributors

Stargazers

Watchers

Forkers

msql-srv's Issues

Recommend Projects

Recommend Topics

Recommend Org