Refactor: Implement OdbcConnection::execute

Co-authored-by: contact <[email protected]>

Author: cursoragent

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

@@ -12,8 +12,8 @@ use odbc_api::Cursor;
 use std::borrow::Cow;
 
 impl OdbcConnection {
-    async fn run<'e>(
-        &'e mut self,
+    async fn run<'e, 'c: 'e>(
+        &'c mut self,
         sql: &'e str,
     ) -> Result<impl futures_core::Stream<Item = Result<Either<OdbcQueryResult, OdbcRow>, Error>> + 'e, Error> {
         let mut logger = QueryLogger::new(sql, self.log_settings.clone());
@@ -71,11 +71,47 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
         'c: 'e,
         E: Execute<'q, Self::Database> + 'q,
     {
-        let sql = query.sql();
+        let sql = query.sql().to_string();
+        let shared = self.worker.shared.clone();
+        let settings = self.log_settings.clone();
         Box::pin(try_stream! {
-            let s = self.run(sql).await?;
-            futures_util::pin_mut!(s);
-            while let Some(v) = s.try_next().await? { r#yield!(v); }
+            let mut logger = QueryLogger::new(&sql, settings.clone());
+            let guard = shared.conn.lock().await;
+            match guard.execute(&sql, (), None) {
+                Ok(Some(mut cursor)) => {
+                    use odbc_api::ResultSetMetadata;
+                    let mut columns = Vec::new();
+                    if let Ok(count) = cursor.num_result_cols() {
+                        for i in 1..=count { // ODBC columns are 1-based
+                            let mut cd = odbc_api::ColumnDescription::default();
+                            let _ = cursor.describe_col(i as u16, &mut cd);
+                            let name = String::from_utf8(cd.name).unwrap_or_else(|_| format!("col{}", i-1));
+                            columns.push(OdbcColumn { name, type_info: OdbcTypeInfo { name: format!("{:?}", cd.data_type), is_null: false }, ordinal: (i-1) as usize });
+                        }
+                    }
+                    while let Some(mut row) = cursor.next_row().map_err(|e| Error::from(e))? {
+                        let mut values = Vec::with_capacity(columns.len());
+                        for i in 1..=columns.len() {
+                            let mut buf = Vec::new();
+                            let not_null = row.get_text(i as u16, &mut buf).map_err(|e| Error::from(e))?;
+                            if not_null {
+                                let ti = OdbcTypeInfo { name: "TEXT".into(), is_null: false };
+                                values.push((ti, Some(buf)));
+                            } else {
+                                let ti = OdbcTypeInfo { name: "TEXT".into(), is_null: true };
+                                values.push((ti, None));
+                            }
+                        }
+                        logger.increment_rows_returned();
+                        r#yield!(Either::Right(OdbcRow { columns: columns.clone(), values }));
+                    }
+                    r#yield!(Either::Left(OdbcQueryResult { rows_affected: 0 }));
+                }
+                Ok(None) => {
+                    r#yield!(Either::Left(OdbcQueryResult { rows_affected: 0 }));
+                }
+                Err(e) => return Err(Error::from(e)),
+            }
             Ok(())
         })
     }