Replace tiberius with mssql (maintained fork with security fixes)

Justfile

@@ -1,6 +1,6 @@
 export POSTGRES_URL := "postgres://user:pass@localhost:5432/db"
 export MYSQL_URL := "mysql://root:pass@localhost:3306/db"
-export TIBERIUS_URL := "tds://sa:passwordA1@localhost:1433"
+export MSSQL_URL := "tds://sa:passwordA1@localhost:1433"
 
 default:
     just --list

README.md

@@ -32,8 +32,8 @@ without need for dynamic linking of C libraries.
 
 | RDBMS | SQLite | DuckDB | PostgreSQL | MySQL | Microsoft SQL Server |
 | --- | --- | --- | --- | --- | --- |
-| feature | `src_rusqlite` | `src_duckdb` | `src_postgres` | `src_mysql` | `src_tiberius` |
-| dependency | [rusqlite](https://crates.io/crates/rusqlite) | [duckdb](https://crates.io/crates/duckdb) | [postgres](https://crates.io/crates/postgres) | [mysql](https://crates.io/crates/mysql) | [tiberius](https://crates.io/crates/tiberius) |
+| feature | `src_rusqlite` | `src_duckdb` | `src_postgres` | `src_mysql` | `src_mssql` |
+| dependency | [rusqlite](https://crates.io/crates/rusqlite) | [duckdb](https://crates.io/crates/duckdb) | [postgres](https://crates.io/crates/postgres) | [mysql](https://crates.io/crates/mysql) | [mssql](https://crates.io/crates/mssql) (fork of [tiberius](https://crates.io/crates/tiberius)) |
 | query | x | x | x | x | x |
 | query params | x | x | x |  | x |
 | schema get | x | x | x | x | x |

connector_arrow/Cargo.toml

@@ -69,8 +69,10 @@ default-features = false
 optional = true
 features = ["minimal"]
 
-[dependencies.tiberius]
-version = "0.12.2"
+[dependencies.mssql]
+# https://crates.io/crates/mssql - a maintained fork of tiberius with the same
+# Client/ColumnData/ToSql/etc. API this module already used from `tiberius`.
+version = "1.0.1"
 default-features = false
 optional = true
 
@@ -91,7 +93,7 @@ all = [
   "src_duckdb",
   "src_postgres",
   "src_mysql",
-  "src_tiberius",
+  "src_mssql",
 ]
 src_postgres = [
   "postgres",
@@ -105,7 +107,7 @@ src_postgres = [
 src_rusqlite = ["rusqlite"]
 src_duckdb = ["duckdb", "fallible-streaming-iterator"]
 src_mysql = ["mysql", "pac_cell"]
-src_tiberius = ["tiberius", "tokio", "tokio-util", "futures"]
+src_mssql = ["mssql", "tokio", "tokio-util", "futures"]
 
 [package.metadata.docs.rs]
 features = ["all"]

connector_arrow/src/errors.rs

@@ -55,9 +55,9 @@ pub enum ConnectorError {
     #[error(transparent)]
     MySQL(#[from] mysql::Error),
 
-    #[cfg(feature = "src_tiberius")]
+    #[cfg(feature = "src_mssql")]
     #[error(transparent)]
-    Tiberius(#[from] tiberius::error::Error),
+    Mssql(#[from] mssql::error::Error),
 }
 
 #[derive(Error, Debug)]

connector_arrow/src/lib.rs

@@ -68,8 +68,8 @@ pub mod mysql;
 pub mod postgres;
 #[cfg(feature = "src_rusqlite")]
 pub mod rusqlite;
-#[cfg(feature = "src_tiberius")]
-pub mod tiberius;
+#[cfg(feature = "src_mssql")]
+pub mod mssql;
 
 pub use arrow;
 pub use errors::*;

connector_arrow/src/tiberius/append.rs → connector_arrow/src/mssql/append.rs

@@ -5,8 +5,8 @@ use arrow::datatypes::*;
 use arrow::record_batch::RecordBatch;
 use futures::{AsyncRead, AsyncWrite};
 use itertools::{zip_eq, Itertools};
-use tiberius::numeric::Numeric;
-use tiberius::{BulkLoadRequest, Client, ColumnData, TokenRow};
+use mssql::numeric::Numeric;
+use mssql::{BulkLoadRequest, Client, ColumnData, TokenRow};
 use tokio::runtime::Runtime;
 
 use crate::api::Append;
@@ -16,20 +16,20 @@ use crate::util::transport::{Consume, ConsumeTy};
 use crate::util::ArrayCellRef;
 use crate::{impl_consume_unsupported, ConnectorError};
 
-pub struct TiberiusAppender<'c, S: AsyncRead + AsyncWrite + Unpin + Send> {
+pub struct MssqlAppender<'c, S: AsyncRead + AsyncWrite + Unpin + Send> {
     rt: Arc<Runtime>,
     bulk_load: BulkLoadRequest<'c, S>,
 }
 
-impl<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> TiberiusAppender<'conn, S> {
+impl<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> MssqlAppender<'conn, S> {
     pub fn new(
         rt: Arc<Runtime>,
         client: &'conn mut Client<S>,
         table_name: &str,
     ) -> Result<Self, ConnectorError> {
         let table_name = escaped_ident(table_name).to_string();
 
-        // Tiberius requires table_name to be 'conn, but does not really use it as such.
+        // mssql (inherited from tiberius) requires table_name to be 'conn, but does not really use it as such.
         // We convert our '_ into 'conn here.
         let table_name: &'conn str = unsafe { std::mem::transmute::<_, _>(table_name.as_str()) };
 
@@ -40,7 +40,7 @@ impl<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> TiberiusAppender<'conn, S>
     }
 }
 
-impl<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> Append<'conn> for TiberiusAppender<'conn, S> {
+impl<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> Append<'conn> for MssqlAppender<'conn, S> {
     fn append(&mut self, batch: RecordBatch) -> Result<(), ConnectorError> {
         let schema = batch.schema();
         let mut row_ref = zip_eq(batch.columns(), schema.fields())

connector_arrow/src/tiberius/mod.rs → connector_arrow/src/mssql/mod.rs

@@ -3,7 +3,7 @@ mod query;
 mod schema;
 mod types;
 
-pub use tiberius;
+pub use mssql;
 
 use arrow::datatypes::*;
 use futures::{AsyncRead, AsyncWrite};
@@ -14,45 +14,45 @@ use tokio::runtime::Runtime;
 use crate::api::Connector;
 use crate::ConnectorError;
 
-pub struct TiberiusConnection<S: AsyncRead + AsyncWrite + Unpin + Send> {
+pub struct MssqlConnection<S: AsyncRead + AsyncWrite + Unpin + Send> {
     rt: Arc<Runtime>,
-    client: tiberius::Client<S>,
+    client: mssql::Client<S>,
 }
 
-impl<S: AsyncRead + AsyncWrite + Unpin + Send> TiberiusConnection<S> {
-    pub fn new(rt: Arc<Runtime>, client: tiberius::Client<S>) -> Self {
-        TiberiusConnection { rt, client }
+impl<S: AsyncRead + AsyncWrite + Unpin + Send> MssqlConnection<S> {
+    pub fn new(rt: Arc<Runtime>, client: mssql::Client<S>) -> Self {
+        MssqlConnection { rt, client }
     }
 
-    pub fn unwrap(self) -> (Arc<Runtime>, tiberius::Client<S>) {
+    pub fn unwrap(self) -> (Arc<Runtime>, mssql::Client<S>) {
         (self.rt, self.client)
     }
 
-    pub fn inner_mut(&mut self) -> (&mut Arc<Runtime>, &mut tiberius::Client<S>) {
+    pub fn inner_mut(&mut self) -> (&mut Arc<Runtime>, &mut mssql::Client<S>) {
         (&mut self.rt, &mut self.client)
     }
 }
 
-impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connector for TiberiusConnection<S> {
+impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connector for MssqlConnection<S> {
     type Stmt<'conn>
-        = query::TiberiusStatement<'conn, S>
+        = query::MssqlStatement<'conn, S>
     where
         Self: 'conn;
 
     type Append<'conn>
-        = append::TiberiusAppender<'conn, S>
+        = append::MssqlAppender<'conn, S>
     where
         Self: 'conn;
 
     fn query<'a>(&'a mut self, query: &str) -> Result<Self::Stmt<'a>, ConnectorError> {
-        Ok(query::TiberiusStatement {
+        Ok(query::MssqlStatement {
             conn: self,
             query: query.to_string(),
         })
     }
 
     fn append<'a>(&'a mut self, table_name: &str) -> Result<Self::Append<'a>, ConnectorError> {
-        append::TiberiusAppender::new(self.rt.clone(), &mut self.client, table_name)
+        append::MssqlAppender::new(self.rt.clone(), &mut self.client, table_name)
     }
 
     #[allow(clippy::get_first)]
@@ -160,7 +160,8 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connector for TiberiusConnection<
 }
 
 fn can_decimal_fit_in_numeric(precision: u8, scale: i8) -> bool {
-    // TODO: this should be p <= 38, not p < 38. This restriction is a bug in tiberius.
+    // TODO: this should be p <= 38, not p < 38. This restriction is a bug inherited from tiberius;
+    // unclear whether the mssql fork has fixed it independently.
 
     precision < 38 && scale >= 0 && precision >= scale as u8
 }

connector_arrow/src/tiberius/query.rs → connector_arrow/src/mssql/query.rs

@@ -2,7 +2,7 @@ use arrow::{datatypes::*, record_batch::RecordBatch};
 use futures::{AsyncRead, AsyncWrite, StreamExt};
 use itertools::Itertools;
 use std::sync::Arc;
-use tiberius::{ColumnData, QueryStream, ToSql};
+use mssql::{ColumnData, QueryStream, ToSql};
 use tokio::runtime::Runtime;
 
 use crate::api::{ResultReader, Statement};
@@ -13,16 +13,16 @@ use crate::util::ArrayCellRef;
 use crate::util::{self, transport::Produce};
 use crate::ConnectorError;
 
-pub struct TiberiusStatement<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> {
-    pub(super) conn: &'conn mut super::TiberiusConnection<S>,
+pub struct MssqlStatement<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> {
+    pub(super) conn: &'conn mut super::MssqlConnection<S>,
     pub(super) query: String,
 }
 
 impl<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> Statement<'conn>
-    for TiberiusStatement<'conn, S>
+    for MssqlStatement<'conn, S>
 {
     type Reader<'stmt>
-        = TiberiusResultReader<'stmt>
+        = MssqlResultReader<'stmt>
     where
         Self: 'stmt;
 
@@ -50,43 +50,43 @@ impl<'conn, S: AsyncRead + AsyncWrite + Unpin + Send> Statement<'conn>
         let schema = super::types::get_result_schema(columns)?;
         self.conn.rt.block_on(stream.next());
 
-        Ok(TiberiusResultReader {
+        Ok(MssqlResultReader {
             schema,
-            stream: TiberiusStream {
+            stream: MssqlStream {
                 rt: self.conn.rt.clone(),
                 stream,
             },
         })
     }
 }
 
-pub struct TiberiusResultReader<'stmt> {
+pub struct MssqlResultReader<'stmt> {
     schema: SchemaRef,
-    stream: TiberiusStream<'stmt>,
+    stream: MssqlStream<'stmt>,
 }
 
-struct TiberiusStream<'stmt> {
+struct MssqlStream<'stmt> {
     rt: Arc<Runtime>,
     stream: QueryStream<'stmt>,
 }
 
-impl<'stmt> ResultReader<'stmt> for TiberiusResultReader<'stmt> {
+impl<'stmt> ResultReader<'stmt> for MssqlResultReader<'stmt> {
     fn get_schema(&mut self) -> Result<arrow::datatypes::SchemaRef, ConnectorError> {
         Ok(self.schema.clone())
     }
 }
 
-impl Iterator for TiberiusResultReader<'_> {
+impl Iterator for MssqlResultReader<'_> {
     type Item = Result<RecordBatch, ConnectorError>;
 
     fn next(&mut self) -> Option<Self::Item> {
         util::next_batch_from_rows(&self.schema, &mut self.stream, 1024).transpose()
     }
 }
 
-impl<'s> util::RowsReader<'s> for TiberiusStream<'s> {
+impl<'s> util::RowsReader<'s> for MssqlStream<'s> {
     type CellReader<'row>
-        = TiberiusCellReader
+        = MssqlCellReader
     where
         Self: 'row;
 
@@ -98,30 +98,30 @@ impl<'s> util::RowsReader<'s> for TiberiusStream<'s> {
 
         // are there more result sets?
         let row = match item? {
-            tiberius::QueryItem::Row(row) => row,
-            tiberius::QueryItem::Metadata(_) => {
+            mssql::QueryItem::Row(row) => row,
+            mssql::QueryItem::Metadata(_) => {
                 // yes, this there are
                 return Err(ConnectorError::MultipleResultSets);
             }
         };
 
-        Ok(Some(TiberiusCellReader { row, cell: 0 }))
+        Ok(Some(MssqlCellReader { row, cell: 0 }))
     }
 }
 
-struct TiberiusCellReader {
-    row: tiberius::Row,
+struct MssqlCellReader {
+    row: mssql::Row,
     cell: usize,
 }
 
-impl util::CellReader<'_> for TiberiusCellReader {
+impl util::CellReader<'_> for MssqlCellReader {
     type CellRef<'cell>
-        = TiberiusCellRef<'cell>
+        = MssqlCellRef<'cell>
     where
         Self: 'cell;
 
     fn next_cell(&mut self) -> Option<Self::CellRef<'_>> {
-        let r = TiberiusCellRef {
+        let r = MssqlCellRef {
             row: &mut self.row,
             cell: self.cell,
         };
@@ -131,20 +131,20 @@ impl util::CellReader<'_> for TiberiusCellReader {
 }
 
 #[derive(Debug)]
-struct TiberiusCellRef<'a> {
-    row: &'a mut tiberius::Row,
+struct MssqlCellRef<'a> {
+    row: &'a mut mssql::Row,
     cell: usize,
 }
 
-impl<'r> Produce<'r> for TiberiusCellRef<'r> {}
+impl<'r> Produce<'r> for MssqlCellRef<'r> {}
 
 macro_rules! impl_produce_ty {
     ($ArrTy: ty, $DbTy: ty) => {
         impl_produce_ty!($ArrTy, $DbTy, std::convert::identity);
     };
 
     ($ArrTy: ty, $DbTy: ty, $conversion: expr) => {
-        impl<'r> ProduceTy<'r, $ArrTy> for TiberiusCellRef<'r> {
+        impl<'r> ProduceTy<'r, $ArrTy> for MssqlCellRef<'r> {
             fn produce(self) -> Result<<$ArrTy as ArrowType>::Native, ConnectorError> {
                 Ok(self
                     .row
@@ -170,7 +170,7 @@ impl_produce_ty!(Utf8Type, StrOrNum, StrOrNum::into_inner);
 impl_produce_ty!(LargeUtf8Type, &str, &str::to_owned);
 
 impl_produce_unsupported!(
-    TiberiusCellRef<'r>,
+    MssqlCellRef<'r>,
     (
         NullType,
         Int8Type,
@@ -211,8 +211,8 @@ impl StrOrNum {
     }
 }
 
-impl<'a> tiberius::FromSql<'a> for StrOrNum {
-    fn from_sql(value: &'a ColumnData<'static>) -> tiberius::Result<Option<Self>> {
+impl<'a> mssql::FromSql<'a> for StrOrNum {
+    fn from_sql(value: &'a ColumnData<'static>) -> mssql::Result<Option<Self>> {
         match value {
             ColumnData::String(s) => Ok(s.as_ref().map(|x| StrOrNum(x.to_string()))),
             ColumnData::Numeric(n) => Ok(n.as_ref().map(|x| {
@@ -228,7 +228,7 @@ impl<'a> tiberius::FromSql<'a> for StrOrNum {
                     StrOrNum(format!("{}", x.value()))
                 }
             })),
-            _ => Err(tiberius::error::Error::Conversion(
+            _ => Err(mssql::error::Error::Conversion(
                 format!("cannot convert `{value:?}` into string").into(),
             )),
         }