Skip to content

Commit 33a0317

Browse files
lovasoalovasoa
committed
feat: Implement ODBC argument encoding and decoding
This commit adds support for encoding and decoding various data types (i32, i64, f32, f64, String, &str, Vec<u8>) for ODBC arguments. It also updates the ODBC connection executor to handle parameterized queries with interpolated SQL, enhancing the query execution capabilities.
1 parent f05b2c2 commit 33a0317

File tree

7 files changed

+378
-12
lines changed

7 files changed

+378
-12
lines changed

‎sqlx-core/src/odbc/arguments.rs‎

Lines changed: 86 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,94 @@ impl<'q> Arguments<'q> for OdbcArguments<'q> {
2525
self.values.reserve(additional);
2626
}
2727

28-
fn add<T>(&mut self, _value: T)
28+
fn add<T>(&mut self, value: T)
2929
where
3030
T: 'q + Send + Encode<'q, Self::Database> + Type<Self::Database>,
3131
{
32-
// Not implemented yet; ODBC backend currently executes direct SQL without binds
33-
// This stub allows query() without binds to compile.
34-
let _ = _value;
32+
let _ = value.encode(&mut self.values);
33+
}
34+
}
35+
36+
impl<'q> Encode<'q, Odbc> for i32 {
37+
fn encode(self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
38+
buf.push(OdbcArgumentValue::Int(self as i64));
39+
crate::encode::IsNull::No
40+
}
41+
42+
fn encode_by_ref(&self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
43+
buf.push(OdbcArgumentValue::Int(*self as i64));
44+
crate::encode::IsNull::No
45+
}
46+
}
47+
48+
impl<'q> Encode<'q, Odbc> for i64 {
49+
fn encode(self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
50+
buf.push(OdbcArgumentValue::Int(self));
51+
crate::encode::IsNull::No
52+
}
53+
54+
fn encode_by_ref(&self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
55+
buf.push(OdbcArgumentValue::Int(*self));
56+
crate::encode::IsNull::No
57+
}
58+
}
59+
60+
impl<'q> Encode<'q, Odbc> for f32 {
61+
fn encode(self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
62+
buf.push(OdbcArgumentValue::Float(self as f64));
63+
crate::encode::IsNull::No
64+
}
65+
66+
fn encode_by_ref(&self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
67+
buf.push(OdbcArgumentValue::Float(*self as f64));
68+
crate::encode::IsNull::No
69+
}
70+
}
71+
72+
impl<'q> Encode<'q, Odbc> for f64 {
73+
fn encode(self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
74+
buf.push(OdbcArgumentValue::Float(self));
75+
crate::encode::IsNull::No
76+
}
77+
78+
fn encode_by_ref(&self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
79+
buf.push(OdbcArgumentValue::Float(*self));
80+
crate::encode::IsNull::No
81+
}
82+
}
83+
84+
impl<'q> Encode<'q, Odbc> for String {
85+
fn encode(self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
86+
buf.push(OdbcArgumentValue::Text(self));
87+
crate::encode::IsNull::No
88+
}
89+
90+
fn encode_by_ref(&self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
91+
buf.push(OdbcArgumentValue::Text(self.clone()));
92+
crate::encode::IsNull::No
93+
}
94+
}
95+
96+
impl<'q> Encode<'q, Odbc> for &'q str {
97+
fn encode(self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
98+
buf.push(OdbcArgumentValue::Text(self.to_owned()));
99+
crate::encode::IsNull::No
100+
}
101+
102+
fn encode_by_ref(&self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
103+
buf.push(OdbcArgumentValue::Text((*self).to_owned()));
104+
crate::encode::IsNull::No
105+
}
106+
}
107+
108+
impl<'q> Encode<'q, Odbc> for Vec<u8> {
109+
fn encode(self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
110+
buf.push(OdbcArgumentValue::Bytes(self));
111+
crate::encode::IsNull::No
112+
}
113+
114+
fn encode_by_ref(&self, buf: &mut Vec<OdbcArgumentValue<'q>>) -> crate::encode::IsNull {
115+
buf.push(OdbcArgumentValue::Bytes(self.clone()));
116+
crate::encode::IsNull::No
35117
}
36118
}

‎sqlx-core/src/odbc/connection/executor.rs‎

Lines changed: 44 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use crate::describe::Describe;
22
use crate::error::Error;
33
use crate::executor::{Execute, Executor};
4-
use crate::odbc::{Odbc, OdbcConnection, OdbcQueryResult, OdbcRow, OdbcStatement, OdbcTypeInfo};
4+
use crate::odbc::{Odbc, OdbcArgumentValue, OdbcConnection, OdbcQueryResult, OdbcRow, OdbcStatement, OdbcTypeInfo};
55
use either::Either;
66
use futures_core::future::BoxFuture;
77
use futures_core::stream::BoxStream;
@@ -15,15 +15,21 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
1515

1616
fn fetch_many<'e, 'q: 'e, E>(
1717
self,
18-
_query: E,
18+
mut _query: E,
1919
) -> BoxStream<'e, Result<Either<OdbcQueryResult, OdbcRow>, Error>>
2020
where
2121
'c: 'e,
2222
E: Execute<'q, Self::Database> + 'q,
2323
{
2424
let sql = _query.sql().to_string();
25+
let mut args = _query.take_arguments();
2526
Box::pin(try_stream! {
26-
let rx = self.worker.execute_stream(&sql).await?;
27+
let rx = if let Some(a) = args.take() {
28+
let new_sql = interpolate_sql_with_odbc_args(&sql, &a.values);
29+
self.worker.execute_stream(&new_sql).await?
30+
} else {
31+
self.worker.execute_stream(&sql).await?
32+
};
2733
while let Ok(item) = rx.recv_async().await {
2834
r#yield!(item?);
2935
}
@@ -76,3 +82,38 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
7682
Box::pin(async move { Err(Error::Protocol("ODBC describe not implemented".into())) })
7783
}
7884
}
85+
86+
fn interpolate_sql_with_odbc_args(sql: &str, args: &[OdbcArgumentValue<'_>]) -> String {
87+
let mut result = String::with_capacity(sql.len() + args.len() * 8);
88+
let mut arg_iter = args.iter();
89+
let mut chars = sql.chars().peekable();
90+
while let Some(ch) = chars.next() {
91+
if ch == '?' {
92+
if let Some(arg) = arg_iter.next() {
93+
match arg {
94+
OdbcArgumentValue::Int(i) => result.push_str(&i.to_string()),
95+
OdbcArgumentValue::Float(f) => result.push_str(&format!("{}", f)),
96+
OdbcArgumentValue::Text(s) => {
97+
result.push('\'');
98+
for c in s.chars() {
99+
if c == '\'' { result.push('\''); }
100+
result.push(c);
101+
}
102+
result.push('\'');
103+
}
104+
OdbcArgumentValue::Bytes(b) => {
105+
result.push_str("X'");
106+
for byte in b { result.push_str(&format!("{:02X}", byte)); }
107+
result.push('\'');
108+
}
109+
OdbcArgumentValue::Null | OdbcArgumentValue::Phantom(_) => result.push_str("NULL"),
110+
}
111+
} else {
112+
result.push('?');
113+
}
114+
} else {
115+
result.push(ch);
116+
}
117+
}
118+
result
119+
}

‎sqlx-core/src/odbc/connection/worker.rs‎

Lines changed: 93 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ use futures_channel::oneshot;
55
use futures_intrusive::sync::Mutex;
66

77
use crate::error::Error;
8-
use crate::odbc::{OdbcColumn, OdbcConnectOptions, OdbcQueryResult, OdbcRow, OdbcTypeInfo};
8+
use crate::odbc::{OdbcArgumentValue, OdbcColumn, OdbcConnectOptions, OdbcQueryResult, OdbcRow, OdbcTypeInfo};
99
use either::Either;
1010
use odbc_api::Cursor;
1111

@@ -40,6 +40,11 @@ enum Command {
4040
sql: Box<str>,
4141
tx: flume::Sender<Result<Either<OdbcQueryResult, OdbcRow>, Error>>,
4242
},
43+
ExecuteWithArgs {
44+
sql: Box<str>,
45+
args: Vec<OdbcArgumentValue<'static>>,
46+
tx: flume::Sender<Result<Either<OdbcQueryResult, OdbcRow>, Error>>,
47+
},
4348
}
4449

4550
impl ConnectionWorker {
@@ -204,6 +209,76 @@ impl ConnectionWorker {
204209
process(&guard);
205210
}
206211
}
212+
Command::ExecuteWithArgs { sql, args, tx } => {
213+
let process = |conn: &odbc_api::Connection<'static>| {
214+
// Fallback: if parameter API is unavailable, execute interpolated SQL directly
215+
match conn.execute(&sql, (), None) {
216+
Ok(Some(mut cursor)) => {
217+
use odbc_api::ResultSetMetadata;
218+
let mut columns: Vec<OdbcColumn> = Vec::new();
219+
if let Ok(count) = cursor.num_result_cols() {
220+
for i in 1..=count {
221+
let mut cd = odbc_api::ColumnDescription::default();
222+
let _ = cursor.describe_col(i as u16, &mut cd);
223+
let name = String::from_utf8(cd.name)
224+
.unwrap_or_else(|_| format!("col{}", i - 1));
225+
columns.push(OdbcColumn {
226+
name,
227+
type_info: OdbcTypeInfo {
228+
name: format!("{:?}", cd.data_type),
229+
is_null: false,
230+
},
231+
ordinal: (i - 1) as usize,
232+
});
233+
}
234+
}
235+
while let Ok(Some(mut row)) = cursor.next_row() {
236+
let mut values: Vec<(OdbcTypeInfo, Option<Vec<u8>>)> =
237+
Vec::with_capacity(columns.len());
238+
for i in 1..=columns.len() {
239+
let mut buf = Vec::new();
240+
// Try text first, then fallback to binary, then numeric
241+
if let Ok(true) = row.get_text(i as u16, &mut buf) {
242+
values.push((OdbcTypeInfo { name: "TEXT".into(), is_null: false }, Some(buf)));
243+
} else if let Ok(false) = row.get_text(i as u16, &mut buf) {
244+
values.push((OdbcTypeInfo { name: "TEXT".into(), is_null: true }, None));
245+
} else if let Ok(bytes) = row.get_binary(i as u16) {
246+
values.push((OdbcTypeInfo { name: "BLOB".into(), is_null: false }, Some(bytes.unwrap_or_default())));
247+
} else if let Ok(opt) = row.get_data::<i64>(i as u16) {
248+
if let Some(num) = opt {
249+
values.push((OdbcTypeInfo { name: "INT".into(), is_null: false }, Some(num.to_string().into_bytes())));
250+
} else {
251+
values.push((OdbcTypeInfo { name: "INT".into(), is_null: true }, None));
252+
}
253+
} else if let Ok(opt) = row.get_data::<f64>(i as u16) {
254+
if let Some(num) = opt {
255+
values.push((OdbcTypeInfo { name: "DOUBLE".into(), is_null: false }, Some(num.to_string().into_bytes())));
256+
} else {
257+
values.push((OdbcTypeInfo { name: "DOUBLE".into(), is_null: true }, None));
258+
}
259+
} else {
260+
values.push((OdbcTypeInfo { name: "UNKNOWN".into(), is_null: true }, None));
261+
}
262+
}
263+
let _ = tx.send(Ok(Either::Right(OdbcRow { columns: columns.clone(), values })));
264+
}
265+
let _ = tx.send(Ok(Either::Left(OdbcQueryResult { rows_affected: 0 })));
266+
}
267+
Ok(None) => {
268+
let _ = tx.send(Ok(Either::Left(OdbcQueryResult { rows_affected: 0 })));
269+
}
270+
Err(e) => {
271+
let _ = tx.send(Err(Error::from(e)));
272+
}
273+
}
274+
};
275+
if let Some(conn) = shared.conn.try_lock() {
276+
process(&conn);
277+
} else {
278+
let guard = futures_executor::block_on(shared.conn.lock());
279+
process(&guard);
280+
}
281+
}
207282
}
208283
}
209284
})?;
@@ -273,4 +348,21 @@ impl ConnectionWorker {
273348
.map_err(|_| Error::WorkerCrashed)?;
274349
Ok(rx)
275350
}
351+
352+
pub(crate) async fn execute_stream_with_args(
353+
&mut self,
354+
sql: &str,
355+
args: Vec<OdbcArgumentValue<'static>>,
356+
) -> Result<flume::Receiver<Result<Either<OdbcQueryResult, OdbcRow>, Error>>, Error> {
357+
let (tx, rx) = flume::bounded(64);
358+
self.command_tx
359+
.send_async(Command::ExecuteWithArgs {
360+
sql: sql.into(),
361+
args,
362+
tx,
363+
})
364+
.await
365+
.map_err(|_| Error::WorkerCrashed)?;
366+
Ok(rx)
367+
}
276368
}

‎sqlx-core/src/odbc/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ mod row;
1313
mod statement;
1414
mod transaction;
1515
mod type_info;
16+
mod r#type;
1617
mod value;
1718

1819
pub use arguments::{OdbcArgumentValue, OdbcArguments};

‎sqlx-core/src/odbc/type.rs‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
use crate::odbc::Odbc;
2+
use crate::types::Type;
3+
use crate::odbc::OdbcTypeInfo;
4+
5+
impl Type<Odbc> for i32 {
6+
fn type_info() -> OdbcTypeInfo { OdbcTypeInfo { name: "INT".into(), is_null: false } }
7+
fn compatible(_ty: &OdbcTypeInfo) -> bool { true }
8+
}
9+
10+
impl Type<Odbc> for i64 {
11+
fn type_info() -> OdbcTypeInfo { OdbcTypeInfo { name: "BIGINT".into(), is_null: false } }
12+
fn compatible(_ty: &OdbcTypeInfo) -> bool { true }
13+
}
14+
15+
impl Type<Odbc> for f64 {
16+
fn type_info() -> OdbcTypeInfo { OdbcTypeInfo { name: "DOUBLE".into(), is_null: false } }
17+
fn compatible(_ty: &OdbcTypeInfo) -> bool { true }
18+
}
19+
20+
impl Type<Odbc> for f32 {
21+
fn type_info() -> OdbcTypeInfo { OdbcTypeInfo { name: "FLOAT".into(), is_null: false } }
22+
fn compatible(_ty: &OdbcTypeInfo) -> bool { true }
23+
}
24+
25+
impl Type<Odbc> for String {
26+
fn type_info() -> OdbcTypeInfo { OdbcTypeInfo { name: "TEXT".into(), is_null: false } }
27+
fn compatible(_ty: &OdbcTypeInfo) -> bool { true }
28+
}
29+
30+
impl<'a> Type<Odbc> for &'a str {
31+
fn type_info() -> OdbcTypeInfo { OdbcTypeInfo { name: "TEXT".into(), is_null: false } }
32+
fn compatible(_ty: &OdbcTypeInfo) -> bool { true }
33+
}
34+
35+
impl Type<Odbc> for Vec<u8> {
36+
fn type_info() -> OdbcTypeInfo { OdbcTypeInfo { name: "BLOB".into(), is_null: false } }
37+
fn compatible(_ty: &OdbcTypeInfo) -> bool { true }
38+
}
39+
40+
// Option<T> blanket impl is provided in core types; do not re-implement here.

0 commit comments

Comments
 (0)