1use std::collections::HashSet;
63use std::sync::{Arc, OnceLock};
64
65use adbc_core::error::{Error as AdbcError, Result as AdbcResult, Status as AdbcStatus};
66use adbc_core::options::{
67 InfoCode, ObjectDepth, OptionConnection, OptionDatabase, OptionStatement, OptionValue,
68};
69use adbc_core::{Optionable, PartitionedResult};
70use arrow::array::{Array, RecordBatch, RecordBatchReader};
71use arrow::compute::concat_batches;
72use arrow::datatypes::Schema;
73use tokio::runtime::Runtime;
74use tokio::sync::Mutex;
75
76use crate::adbc::Connection as ExaConnection;
77use crate::error::{ExasolError, QueryError};
78use crate::query::prepared::PreparedStatement;
79use crate::query::Parameter;
80use crate::transport::messages::DataType as TransportDataType;
81use crate::types::{ExasolType, TypeMapper};
82
83fn get_runtime() -> &'static Runtime {
88 static RUNTIME: OnceLock<Runtime> = OnceLock::new();
89 RUNTIME.get_or_init(|| {
90 tokio::runtime::Builder::new_multi_thread()
91 .worker_threads(2)
92 .enable_all()
93 .build()
94 .expect("Failed to create tokio runtime for ADBC FFI")
95 })
96}
97
98fn to_adbc_error(err: impl std::error::Error) -> AdbcError {
100 AdbcError::with_message_and_status(err.to_string(), AdbcStatus::Internal)
101}
102
103type ColumnMetadataMap = std::collections::HashMap<(String, String), Vec<(String, i32, String)>>;
105
106fn quote_identifier(name: &str) -> String {
108 format!("\"{}\"", name.replace('"', "\"\""))
109}
110
111fn build_qualified_table_name(schema: Option<&str>, table: &str) -> String {
118 if let Some(schema) = schema {
119 format!("{}.{}", quote_identifier(schema), quote_identifier(table))
120 } else if let Some((schema_part, table_part)) = table.split_once('.') {
121 format!(
122 "{}.{}",
123 quote_identifier(schema_part),
124 quote_identifier(table_part)
125 )
126 } else {
127 quote_identifier(table)
128 }
129}
130
131pub(crate) fn generate_create_table_ddl(table_name: &str, schema: &Schema) -> AdbcResult<String> {
136 let mut columns = Vec::new();
137
138 for field in schema.fields() {
139 let exasol_type = TypeMapper::arrow_to_exasol(field.data_type()).map_err(|e| {
140 AdbcError::with_message_and_status(
141 format!(
142 "Cannot map Arrow type {:?} for column '{}': {}",
143 field.data_type(),
144 field.name(),
145 e
146 ),
147 AdbcStatus::InvalidArguments,
148 )
149 })?;
150
151 let ddl_type = exasol_type.to_ddl_type();
152 let escaped_name = format!("\"{}\"", field.name().replace('"', "\"\""));
153
154 let col_def = if !field.is_nullable() {
155 format!("{} {} NOT NULL", escaped_name, ddl_type)
156 } else {
157 format!("{} {}", escaped_name, ddl_type)
158 };
159 columns.push(col_def);
160 }
161
162 Ok(format!(
163 "CREATE TABLE {} ({})",
164 table_name,
165 columns.join(", ")
166 ))
167}
168
169fn parse_exasol_type_string(type_str: &str) -> AdbcResult<ExasolType> {
174 let type_upper = type_str.trim().to_uppercase();
175
176 if type_upper == "BOOLEAN" {
177 return Ok(ExasolType::Boolean);
178 }
179 if type_upper == "DATE" {
180 return Ok(ExasolType::Date);
181 }
182 if type_upper == "DOUBLE" || type_upper == "DOUBLE PRECISION" {
183 return Ok(ExasolType::Double);
184 }
185 if type_upper.starts_with("TIMESTAMP") && type_upper.ends_with("WITH LOCAL TIME ZONE") {
186 return Ok(ExasolType::Timestamp {
187 with_local_time_zone: true,
188 });
189 }
190 if type_upper.starts_with("TIMESTAMP") {
191 return Ok(ExasolType::Timestamp {
192 with_local_time_zone: false,
193 });
194 }
195 if type_upper.starts_with("INTERVAL YEAR") && type_upper.contains("TO MONTH") {
196 return Ok(ExasolType::IntervalYearToMonth);
197 }
198 if type_upper.starts_with("INTERVAL DAY") && type_upper.contains("TO SECOND") {
199 let precision = extract_last_param(&type_upper).unwrap_or(3);
200 return Ok(ExasolType::IntervalDayToSecond {
201 precision: precision as u8,
202 });
203 }
204 if type_upper.starts_with("GEOMETRY") {
205 let srid = extract_single_param(&type_upper).map(|v| v as i32);
206 return Ok(ExasolType::Geometry { srid });
207 }
208 if type_upper.starts_with("HASHTYPE") {
209 let byte_size = extract_single_param(&type_upper).unwrap_or(16);
210 return Ok(ExasolType::Hashtype {
211 byte_size: byte_size as usize,
212 });
213 }
214 if type_upper.starts_with("CHAR") && !type_upper.starts_with("CHAR VARYING") {
215 let size = extract_single_param(&type_upper).unwrap_or(1);
216 return Ok(ExasolType::Char {
217 size: size as usize,
218 });
219 }
220 if type_upper.starts_with("VARCHAR") || type_upper.starts_with("CHAR VARYING") {
221 let size = extract_single_param(&type_upper).unwrap_or(2000000);
222 return Ok(ExasolType::Varchar {
223 size: size as usize,
224 });
225 }
226 if type_upper.starts_with("DECIMAL") || type_upper.starts_with("NUMERIC") {
227 let (p, s) = extract_two_params(&type_upper).unwrap_or((18, 0));
228 return Ok(ExasolType::Decimal {
229 precision: p as u8,
230 scale: s as i8,
231 });
232 }
233 if type_upper == "BIGINT"
234 || type_upper == "INT"
235 || type_upper == "INTEGER"
236 || type_upper == "SMALLINT"
237 || type_upper == "TINYINT"
238 {
239 return Ok(ExasolType::Decimal {
240 precision: 18,
241 scale: 0,
242 });
243 }
244 if type_upper == "FLOAT" || type_upper == "REAL" {
245 return Ok(ExasolType::Double);
246 }
247 if type_upper == "CLOB" || type_upper.starts_with("LONG VARCHAR") {
248 return Ok(ExasolType::Varchar { size: 2000000 });
249 }
250
251 Err(AdbcError::with_message_and_status(
252 format!("Unknown Exasol type: {}", type_str),
253 AdbcStatus::Internal,
254 ))
255}
256
257fn extract_single_param(s: &str) -> Option<i64> {
258 let start = s.find('(')?;
259 let end = s.find(')')?;
260 s[start + 1..end].trim().parse().ok()
261}
262
263fn extract_last_param(s: &str) -> Option<i64> {
264 let start = s.rfind('(')?;
265 let end = s.rfind(')')?;
266 if start < end {
267 s[start + 1..end].trim().parse().ok()
268 } else {
269 None
270 }
271}
272
273fn extract_two_params(s: &str) -> Option<(i64, i64)> {
274 let start = s.find('(')?;
275 let end = s.find(')')?;
276 let inner = &s[start + 1..end];
277 let parts: Vec<&str> = inner.split(',').collect();
278 if parts.len() == 2 {
279 let a = parts[0].trim().parse().ok()?;
280 let b = parts[1].trim().parse().ok()?;
281 Some((a, b))
282 } else if parts.len() == 1 {
283 let a = parts[0].trim().parse().ok()?;
284 Some((a, 0))
285 } else {
286 None
287 }
288}
289
290fn transport_datatype_to_exasol(dt: &TransportDataType) -> AdbcResult<ExasolType> {
292 let type_upper = dt.type_name.to_uppercase();
293
294 match type_upper.as_str() {
295 "BOOLEAN" => Ok(ExasolType::Boolean),
296 "DATE" => Ok(ExasolType::Date),
297 "DOUBLE" | "DOUBLE PRECISION" => Ok(ExasolType::Double),
298 "TIMESTAMP" => Ok(ExasolType::Timestamp {
299 with_local_time_zone: dt.with_local_time_zone.unwrap_or(false),
300 }),
301 "TIMESTAMP WITH LOCAL TIME ZONE" => Ok(ExasolType::Timestamp {
302 with_local_time_zone: true,
303 }),
304 "CHAR" => Ok(ExasolType::Char {
305 size: dt.size.unwrap_or(1) as usize,
306 }),
307 "VARCHAR" => Ok(ExasolType::Varchar {
308 size: dt.size.unwrap_or(2000000) as usize,
309 }),
310 "DECIMAL" => {
311 let precision = dt.precision.unwrap_or(18) as u8;
312 let scale = dt.scale.unwrap_or(0) as i8;
313 Ok(ExasolType::Decimal { precision, scale })
314 }
315 "GEOMETRY" => Ok(ExasolType::Geometry { srid: None }),
316 "HASHTYPE" => Ok(ExasolType::Hashtype {
317 byte_size: dt.size.unwrap_or(16) as usize,
318 }),
319 "INTERVAL DAY TO SECOND" => Ok(ExasolType::IntervalDayToSecond {
320 precision: dt.fraction.unwrap_or(3) as u8,
321 }),
322 "INTERVAL YEAR TO MONTH" => Ok(ExasolType::IntervalYearToMonth),
323 _ => Err(AdbcError::with_message_and_status(
324 format!("Unknown transport type: {}", dt.type_name),
325 AdbcStatus::Internal,
326 )),
327 }
328}
329
330fn build_get_objects_schema() -> Schema {
332 use arrow::datatypes::{DataType, Field, Fields};
333
334 let column_fields = Fields::from(vec![
335 Field::new("column_name", DataType::Utf8, false),
336 Field::new("ordinal_position", DataType::Int32, false),
337 Field::new("xdbc_type_name", DataType::Utf8, true),
338 ]);
339
340 let constraint_fields = Fields::from(vec![
341 Field::new("constraint_name", DataType::Utf8, true),
342 Field::new("constraint_type", DataType::Utf8, false),
343 ]);
344
345 let table_fields = Fields::from(vec![
346 Field::new("table_name", DataType::Utf8, false),
347 Field::new("table_type", DataType::Utf8, false),
348 Field::new(
349 "table_columns",
350 DataType::List(Arc::new(Field::new(
351 "item",
352 DataType::Struct(column_fields),
353 true,
354 ))),
355 true,
356 ),
357 Field::new(
358 "table_constraints",
359 DataType::List(Arc::new(Field::new(
360 "item",
361 DataType::Struct(constraint_fields),
362 true,
363 ))),
364 true,
365 ),
366 ]);
367
368 let schema_fields = Fields::from(vec![
369 Field::new("db_schema_name", DataType::Utf8, true),
370 Field::new(
371 "db_schema_tables",
372 DataType::List(Arc::new(Field::new(
373 "item",
374 DataType::Struct(table_fields),
375 true,
376 ))),
377 true,
378 ),
379 ]);
380
381 Schema::new(vec![
382 Field::new("catalog_name", DataType::Utf8, false),
383 Field::new(
384 "catalog_db_schemas",
385 DataType::List(Arc::new(Field::new(
386 "item",
387 DataType::Struct(schema_fields),
388 true,
389 ))),
390 true,
391 ),
392 ])
393}
394
395fn get_int_value(array: &dyn arrow::array::Array, idx: usize) -> Option<i32> {
398 use arrow::array::{Float64Array, Int32Array, Int64Array};
399 use arrow::datatypes::DataType;
400
401 match array.data_type() {
402 DataType::Int32 => {
403 let arr = array.as_any().downcast_ref::<Int32Array>()?;
404 if arr.is_null(idx) {
405 None
406 } else {
407 Some(arr.value(idx))
408 }
409 }
410 DataType::Int64 => {
411 let arr = array.as_any().downcast_ref::<Int64Array>()?;
412 if arr.is_null(idx) {
413 None
414 } else {
415 Some(arr.value(idx) as i32)
416 }
417 }
418 DataType::Float64 => {
419 let arr = array.as_any().downcast_ref::<Float64Array>()?;
420 if arr.is_null(idx) {
421 None
422 } else {
423 Some(arr.value(idx) as i32)
424 }
425 }
426 DataType::Decimal128(_, _) => {
427 let arr = array
428 .as_any()
429 .downcast_ref::<arrow::array::Decimal128Array>()?;
430 if arr.is_null(idx) {
431 None
432 } else {
433 Some(arr.value(idx) as i32)
434 }
435 }
436 _ => None,
437 }
438}
439
440fn get_nullable_value(array: &dyn arrow::array::Array, idx: usize) -> bool {
443 use arrow::array::{BooleanArray, StringArray};
444 use arrow::datatypes::DataType;
445
446 match array.data_type() {
447 DataType::Boolean => {
448 if let Some(arr) = array.as_any().downcast_ref::<BooleanArray>() {
449 if arr.is_null(idx) {
450 true
451 } else {
452 arr.value(idx)
453 }
454 } else {
455 true
456 }
457 }
458 DataType::Utf8 => {
459 if let Some(arr) = array.as_any().downcast_ref::<StringArray>() {
460 if arr.is_null(idx) {
461 true
462 } else {
463 let val = arr.value(idx).to_uppercase();
464 val == "TRUE" || val == "YES" || val == "1"
465 }
466 } else {
467 true
468 }
469 }
470 _ => true,
471 }
472}
473
474#[derive(Debug, Default)]
486pub struct FfiDriver;
487
488impl adbc_core::Driver for FfiDriver {
489 type DatabaseType = FfiDatabase;
490
491 fn new_database(&mut self) -> AdbcResult<Self::DatabaseType> {
492 Ok(FfiDatabase::new())
493 }
494
495 fn new_database_with_opts(
496 &mut self,
497 opts: impl IntoIterator<Item = (OptionDatabase, OptionValue)>,
498 ) -> AdbcResult<Self::DatabaseType> {
499 let mut db = FfiDatabase::new();
500 for (key, value) in opts {
501 db.set_option(key, value)?;
502 }
503 Ok(db)
504 }
505}
506
507pub struct FfiDatabase {
523 uri: Option<String>,
525 username: Option<String>,
527 password: Option<String>,
529 options: std::collections::HashMap<String, OptionValue>,
531}
532
533impl FfiDatabase {
534 fn new() -> Self {
535 Self {
536 uri: None,
537 username: None,
538 password: None,
539 options: std::collections::HashMap::new(),
540 }
541 }
542
543 fn build_connection_uri(&self) -> AdbcResult<String> {
546 let uri = self.uri.as_ref().ok_or_else(|| {
547 AdbcError::with_message_and_status(
548 "Database URI not set. Set adbc.exasol.uri option.",
549 AdbcStatus::InvalidState,
550 )
551 })?;
552
553 if self.username.is_none() && self.password.is_none() {
555 return Ok(uri.clone());
556 }
557
558 let uri_str = uri.as_str();
561 if !uri_str.starts_with("exasol://") {
562 return Err(AdbcError::with_message_and_status(
563 "URI must start with exasol://",
564 AdbcStatus::InvalidArguments,
565 ));
566 }
567
568 let after_scheme = &uri_str[9..]; let at_pos = after_scheme.rfind('@');
572 let (host_part, orig_user, orig_pass) = if let Some(at) = at_pos {
573 let auth_part = &after_scheme[..at];
574 let host_part = &after_scheme[at + 1..];
575 let (user, pass) = if let Some(colon) = auth_part.find(':') {
576 (&auth_part[..colon], Some(&auth_part[colon + 1..]))
577 } else {
578 (auth_part, None)
579 };
580 (host_part, Some(user), pass)
581 } else {
582 (after_scheme, None, None)
583 };
584
585 let user = self.username.as_deref().or(orig_user).unwrap_or("sys");
587 let pass = self.password.as_deref().or(orig_pass).unwrap_or("");
588
589 if pass.is_empty() {
591 Ok(format!("exasol://{}@{}", user, host_part))
592 } else {
593 Ok(format!("exasol://{}:{}@{}", user, pass, host_part))
594 }
595 }
596}
597
598impl Optionable for FfiDatabase {
599 type Option = OptionDatabase;
600
601 fn set_option(&mut self, key: Self::Option, value: OptionValue) -> AdbcResult<()> {
602 match key {
603 OptionDatabase::Uri => {
604 if let OptionValue::String(s) = value {
605 self.uri = Some(s);
606 } else {
607 return Err(AdbcError::with_message_and_status(
608 "URI must be a string",
609 AdbcStatus::InvalidArguments,
610 ));
611 }
612 }
613 OptionDatabase::Username => {
614 if let OptionValue::String(s) = value {
615 self.username = Some(s);
616 } else {
617 return Err(AdbcError::with_message_and_status(
618 "Username must be a string",
619 AdbcStatus::InvalidArguments,
620 ));
621 }
622 }
623 OptionDatabase::Password => {
624 if let OptionValue::String(s) = value {
625 self.password = Some(s);
626 } else {
627 return Err(AdbcError::with_message_and_status(
628 "Password must be a string",
629 AdbcStatus::InvalidArguments,
630 ));
631 }
632 }
633 OptionDatabase::Other(key) => {
634 self.options.insert(key, value);
635 }
636 _ => {
637 return Err(AdbcError::with_message_and_status(
639 "Unsupported database option",
640 AdbcStatus::NotImplemented,
641 ));
642 }
643 }
644 Ok(())
645 }
646
647 fn get_option_string(&self, key: Self::Option) -> AdbcResult<String> {
648 match key {
649 OptionDatabase::Uri => self.uri.clone().ok_or_else(|| {
650 AdbcError::with_message_and_status("URI not set", AdbcStatus::NotFound)
651 }),
652 OptionDatabase::Username => self.username.clone().ok_or_else(|| {
653 AdbcError::with_message_and_status("Username not set", AdbcStatus::NotFound)
654 }),
655 OptionDatabase::Password => Err(AdbcError::with_message_and_status(
656 "Password cannot be retrieved",
657 AdbcStatus::InvalidArguments,
658 )),
659 OptionDatabase::Other(key) => {
660 if let Some(OptionValue::String(s)) = self.options.get(&key) {
661 Ok(s.clone())
662 } else {
663 Err(AdbcError::with_message_and_status(
664 format!("Option {} not found or not a string", key),
665 AdbcStatus::NotFound,
666 ))
667 }
668 }
669 _ => Err(AdbcError::with_message_and_status(
670 "Option not found",
671 AdbcStatus::NotFound,
672 )),
673 }
674 }
675
676 fn get_option_bytes(&self, key: Self::Option) -> AdbcResult<Vec<u8>> {
677 if let OptionDatabase::Other(key) = key {
678 if let Some(OptionValue::Bytes(b)) = self.options.get(&key) {
679 return Ok(b.clone());
680 }
681 }
682 Err(AdbcError::with_message_and_status(
683 "Option not found or not bytes",
684 AdbcStatus::NotFound,
685 ))
686 }
687
688 fn get_option_int(&self, key: Self::Option) -> AdbcResult<i64> {
689 if let OptionDatabase::Other(key) = key {
690 if let Some(OptionValue::Int(i)) = self.options.get(&key) {
691 return Ok(*i);
692 }
693 }
694 Err(AdbcError::with_message_and_status(
695 "Option not found or not an integer",
696 AdbcStatus::NotFound,
697 ))
698 }
699
700 fn get_option_double(&self, key: Self::Option) -> AdbcResult<f64> {
701 if let OptionDatabase::Other(key) = key {
702 if let Some(OptionValue::Double(d)) = self.options.get(&key) {
703 return Ok(*d);
704 }
705 }
706 Err(AdbcError::with_message_and_status(
707 "Option not found or not a double",
708 AdbcStatus::NotFound,
709 ))
710 }
711}
712
713impl adbc_core::Database for FfiDatabase {
714 type ConnectionType = FfiConnection;
715
716 fn new_connection(&self) -> AdbcResult<Self::ConnectionType> {
717 let uri = self.build_connection_uri()?;
718 Ok(FfiConnection::new(uri))
719 }
720
721 fn new_connection_with_opts(
722 &self,
723 opts: impl IntoIterator<Item = (OptionConnection, OptionValue)>,
724 ) -> AdbcResult<Self::ConnectionType> {
725 let uri = self.build_connection_uri()?;
726 let mut conn = FfiConnection::new(uri);
727 for (key, value) in opts {
728 conn.set_option(key, value)?;
729 }
730 Ok(conn)
731 }
732}
733
734pub struct FfiConnection {
744 uri: String,
746 inner: Option<Arc<Mutex<ExaConnection>>>,
748 options: std::collections::HashMap<String, OptionValue>,
750 auto_commit: bool,
752 current_schema: Option<String>,
754}
755
756impl FfiConnection {
757 fn new(uri: String) -> Self {
758 Self {
759 uri,
760 inner: None,
761 options: std::collections::HashMap::new(),
762 auto_commit: true,
763 current_schema: None,
764 }
765 }
766
767 fn ensure_connected(&mut self) -> AdbcResult<Arc<Mutex<ExaConnection>>> {
769 if self.inner.is_none() {
770 let uri = self.uri.clone();
771 let conn = get_runtime()
772 .block_on(async {
773 let params: crate::connection::ConnectionParams = uri.parse()?;
774 ExaConnection::from_params(params).await
775 })
776 .map_err(to_adbc_error)?;
777 self.inner = Some(Arc::new(Mutex::new(conn)));
778 }
779 Ok(Arc::clone(self.inner.as_ref().unwrap()))
780 }
781}
782
783impl Drop for FfiConnection {
784 fn drop(&mut self) {
785 if let Some(conn) = self.inner.take() {
786 let _ = get_runtime().block_on(async {
789 let conn = conn.lock().await;
790 conn.shutdown().await
791 });
792 }
793 }
794}
795
796impl Optionable for FfiConnection {
797 type Option = OptionConnection;
798
799 fn set_option(&mut self, key: Self::Option, value: OptionValue) -> AdbcResult<()> {
800 match key {
801 OptionConnection::AutoCommit => {
802 if let OptionValue::String(s) = value {
803 let new_auto_commit = s == "true" || s == "1";
804 let old_auto_commit = self.auto_commit;
805 self.auto_commit = new_auto_commit;
806
807 if old_auto_commit && !new_auto_commit {
808 let conn_arc = self.ensure_connected()?;
809 get_runtime()
810 .block_on(async {
811 let mut conn = conn_arc.lock().await;
812 conn.begin_transaction().await
813 })
814 .map_err(to_adbc_error)?;
815 } else if !old_auto_commit && new_auto_commit {
816 let conn_arc = self.ensure_connected()?;
817 get_runtime()
818 .block_on(async {
819 let mut conn = conn_arc.lock().await;
820 if conn.in_transaction() {
821 conn.commit().await?;
822 }
823 Ok::<(), QueryError>(())
824 })
825 .map_err(to_adbc_error)?;
826 }
827 } else {
828 return Err(AdbcError::with_message_and_status(
829 "AutoCommit must be a string ('true' or 'false')",
830 AdbcStatus::InvalidArguments,
831 ));
832 }
833 }
834 OptionConnection::CurrentSchema => {
835 if let OptionValue::String(s) = value {
836 self.current_schema = Some(s);
837 } else {
838 return Err(AdbcError::with_message_and_status(
839 "CurrentSchema must be a string",
840 AdbcStatus::InvalidArguments,
841 ));
842 }
843 }
844 OptionConnection::ReadOnly | OptionConnection::IsolationLevel => {
845 self.options.insert(key.as_ref().to_string(), value);
847 }
848 OptionConnection::CurrentCatalog => {
849 return Err(AdbcError::with_message_and_status(
851 "Exasol does not support catalogs",
852 AdbcStatus::NotImplemented,
853 ));
854 }
855 OptionConnection::Other(key) => {
856 self.options.insert(key, value);
857 }
858 _ => {
859 return Err(AdbcError::with_message_and_status(
861 "Unsupported connection option",
862 AdbcStatus::NotImplemented,
863 ));
864 }
865 }
866 Ok(())
867 }
868
869 fn get_option_string(&self, key: Self::Option) -> AdbcResult<String> {
870 match key {
871 OptionConnection::AutoCommit => {
872 Ok(if self.auto_commit { "true" } else { "false" }.to_string())
873 }
874 OptionConnection::CurrentCatalog => Ok("EXA".to_string()),
875 OptionConnection::CurrentSchema => self.current_schema.clone().ok_or_else(|| {
876 AdbcError::with_message_and_status("CurrentSchema not set", AdbcStatus::NotFound)
877 }),
878 OptionConnection::Other(key) => {
879 if let Some(OptionValue::String(s)) = self.options.get(&key) {
880 Ok(s.clone())
881 } else {
882 Err(AdbcError::with_message_and_status(
883 format!("Option {} not found", key),
884 AdbcStatus::NotFound,
885 ))
886 }
887 }
888 _ => Err(AdbcError::with_message_and_status(
889 "Option not found",
890 AdbcStatus::NotFound,
891 )),
892 }
893 }
894
895 fn get_option_bytes(&self, key: Self::Option) -> AdbcResult<Vec<u8>> {
896 if let OptionConnection::Other(key) = key {
897 if let Some(OptionValue::Bytes(b)) = self.options.get(&key) {
898 return Ok(b.clone());
899 }
900 }
901 Err(AdbcError::with_message_and_status(
902 "Option not found or not bytes",
903 AdbcStatus::NotFound,
904 ))
905 }
906
907 fn get_option_int(&self, key: Self::Option) -> AdbcResult<i64> {
908 if let OptionConnection::Other(key) = key {
909 if let Some(OptionValue::Int(i)) = self.options.get(&key) {
910 return Ok(*i);
911 }
912 }
913 Err(AdbcError::with_message_and_status(
914 "Option not found or not an integer",
915 AdbcStatus::NotFound,
916 ))
917 }
918
919 fn get_option_double(&self, key: Self::Option) -> AdbcResult<f64> {
920 if let OptionConnection::Other(key) = key {
921 if let Some(OptionValue::Double(d)) = self.options.get(&key) {
922 return Ok(*d);
923 }
924 }
925 Err(AdbcError::with_message_and_status(
926 "Option not found or not a double",
927 AdbcStatus::NotFound,
928 ))
929 }
930}
931
932struct VecRecordBatchReader {
934 schema: Arc<Schema>,
935 batches: std::vec::IntoIter<RecordBatch>,
936}
937
938impl VecRecordBatchReader {
939 fn new(schema: Arc<Schema>, batches: Vec<RecordBatch>) -> Self {
940 Self {
941 schema,
942 batches: batches.into_iter(),
943 }
944 }
945
946 fn empty(schema: Arc<Schema>) -> Self {
947 Self::new(schema, vec![])
948 }
949}
950
951impl Iterator for VecRecordBatchReader {
952 type Item = Result<RecordBatch, arrow::error::ArrowError>;
953
954 fn next(&mut self) -> Option<Self::Item> {
955 self.batches.next().map(Ok)
956 }
957}
958
959impl RecordBatchReader for VecRecordBatchReader {
960 fn schema(&self) -> Arc<Schema> {
961 Arc::clone(&self.schema)
962 }
963}
964
965impl adbc_core::Connection for FfiConnection {
966 type StatementType = FfiStatement;
967
968 fn new_statement(&mut self) -> AdbcResult<Self::StatementType> {
969 let conn = self.ensure_connected()?;
970 let mut stmt = FfiStatement::with_connection(Some(conn), self.uri.clone());
971 stmt.auto_commit = self.auto_commit;
972 Ok(stmt)
973 }
974
975 fn cancel(&mut self) -> AdbcResult<()> {
976 Err(AdbcError::with_message_and_status(
978 "Cancel not implemented",
979 AdbcStatus::NotImplemented,
980 ))
981 }
982
983 fn get_info(
984 &self,
985 _codes: Option<HashSet<InfoCode>>,
986 ) -> AdbcResult<Box<dyn RecordBatchReader + Send>> {
987 use arrow::array::builder::{StringBuilder, UInt32Builder};
989 use arrow::datatypes::{DataType, Field};
990
991 let schema = Arc::new(Schema::new(vec![
993 Field::new("info_name", DataType::UInt32, false),
994 Field::new("info_value", DataType::Utf8, true),
995 ]));
996
997 let mut name_builder = UInt32Builder::new();
999 let mut value_builder = StringBuilder::new();
1000
1001 name_builder.append_value(0); value_builder.append_value("Exasol");
1004
1005 name_builder.append_value(100); value_builder.append_value("exarrow-rs");
1007
1008 name_builder.append_value(101); value_builder.append_value(env!("CARGO_PKG_VERSION"));
1010
1011 let batch = RecordBatch::try_new(
1012 Arc::clone(&schema),
1013 vec![
1014 Arc::new(name_builder.finish()),
1015 Arc::new(value_builder.finish()),
1016 ],
1017 )
1018 .map_err(|e| AdbcError::with_message_and_status(e.to_string(), AdbcStatus::Internal))?;
1019
1020 Ok(Box::new(VecRecordBatchReader::new(schema, vec![batch])))
1021 }
1022
1023 fn get_objects(
1024 &self,
1025 depth: ObjectDepth,
1026 catalog: Option<&str>,
1027 db_schema: Option<&str>,
1028 table_name: Option<&str>,
1029 table_type: Option<Vec<&str>>,
1030 column_name: Option<&str>,
1031 ) -> AdbcResult<Box<dyn RecordBatchReader + Send>> {
1032 use arrow::array::builder::{
1033 ArrayBuilder, Int32Builder, ListBuilder, StringBuilder, StructBuilder,
1034 };
1035 use arrow::datatypes::{DataType, Field, Fields};
1036
1037 if let Some(cat) = catalog {
1039 if !cat.eq_ignore_ascii_case("EXA") {
1040 let schema = Arc::new(build_get_objects_schema());
1042 return Ok(Box::new(VecRecordBatchReader::empty(schema)));
1043 }
1044 }
1045
1046 let conn_arc = self.inner.as_ref().ok_or_else(|| {
1047 AdbcError::with_message_and_status(
1048 "Connection not established. Call new_statement() or connect first.",
1049 AdbcStatus::InvalidState,
1050 )
1051 })?;
1052 let conn_arc = Arc::clone(conn_arc);
1053
1054 let include_schemas = matches!(
1056 depth,
1057 ObjectDepth::Schemas | ObjectDepth::Tables | ObjectDepth::Columns | ObjectDepth::All
1058 );
1059 let include_tables = matches!(
1060 depth,
1061 ObjectDepth::Tables | ObjectDepth::Columns | ObjectDepth::All
1062 );
1063 let include_columns = matches!(depth, ObjectDepth::Columns | ObjectDepth::All);
1064
1065 let schemas: Vec<String> = if include_schemas {
1067 let mut sql = "SELECT SCHEMA_NAME FROM SYS.EXA_ALL_SCHEMAS".to_string();
1068 if let Some(schema_filter) = db_schema {
1069 sql.push_str(&format!(
1070 " WHERE SCHEMA_NAME LIKE '{}'",
1071 schema_filter.replace('\'', "''")
1072 ));
1073 }
1074 sql.push_str(" ORDER BY SCHEMA_NAME");
1075
1076 let batches = {
1077 let conn_arc = Arc::clone(&conn_arc);
1078 get_runtime()
1079 .block_on(async {
1080 let mut conn = conn_arc.lock().await;
1081 conn.query(&sql).await
1082 })
1083 .map_err(to_adbc_error)?
1084 };
1085
1086 let mut result = Vec::new();
1087 for batch in &batches {
1088 let col = batch
1089 .column(0)
1090 .as_any()
1091 .downcast_ref::<arrow::array::StringArray>()
1092 .ok_or_else(|| {
1093 AdbcError::with_message_and_status(
1094 "Expected string column for SCHEMA_NAME",
1095 AdbcStatus::Internal,
1096 )
1097 })?;
1098 for i in 0..col.len() {
1099 if !col.is_null(i) {
1100 result.push(col.value(i).to_string());
1101 }
1102 }
1103 }
1104 result
1105 } else {
1106 vec![]
1107 };
1108
1109 let tables: std::collections::HashMap<String, Vec<(String, String)>> = if include_tables {
1111 let mut sql =
1112 "SELECT OBJECT_NAME, OBJECT_TYPE, ROOT_NAME FROM SYS.EXA_ALL_OBJECTS".to_string();
1113 let mut conditions = vec!["OBJECT_TYPE IN ('TABLE', 'VIEW')".to_string()];
1114 if let Some(schema_filter) = db_schema {
1115 conditions.push(format!(
1116 "ROOT_NAME LIKE '{}'",
1117 schema_filter.replace('\'', "''")
1118 ));
1119 }
1120 if let Some(tbl_name) = table_name {
1121 conditions.push(format!(
1122 "OBJECT_NAME LIKE '{}'",
1123 tbl_name.replace('\'', "''")
1124 ));
1125 }
1126 if let Some(ref tbl_types) = table_type {
1127 let types_str: Vec<String> = tbl_types
1128 .iter()
1129 .map(|t| format!("'{}'", t.replace('\'', "''")))
1130 .collect();
1131 conditions.push(format!("OBJECT_TYPE IN ({})", types_str.join(",")));
1132 }
1133 if !conditions.is_empty() {
1134 sql.push_str(" WHERE ");
1135 sql.push_str(&conditions.join(" AND "));
1136 }
1137 sql.push_str(" ORDER BY ROOT_NAME, OBJECT_NAME");
1138
1139 let batches = {
1140 let conn_arc = Arc::clone(&conn_arc);
1141 get_runtime()
1142 .block_on(async {
1143 let mut conn = conn_arc.lock().await;
1144 conn.query(&sql).await
1145 })
1146 .map_err(to_adbc_error)?
1147 };
1148
1149 let mut result: std::collections::HashMap<String, Vec<(String, String)>> =
1150 std::collections::HashMap::new();
1151 for batch in &batches {
1152 let name_col = batch
1153 .column(0)
1154 .as_any()
1155 .downcast_ref::<arrow::array::StringArray>()
1156 .ok_or_else(|| {
1157 AdbcError::with_message_and_status(
1158 "Expected string column",
1159 AdbcStatus::Internal,
1160 )
1161 })?;
1162 let type_col = batch
1163 .column(1)
1164 .as_any()
1165 .downcast_ref::<arrow::array::StringArray>()
1166 .ok_or_else(|| {
1167 AdbcError::with_message_and_status(
1168 "Expected string column",
1169 AdbcStatus::Internal,
1170 )
1171 })?;
1172 let schema_col = batch
1173 .column(2)
1174 .as_any()
1175 .downcast_ref::<arrow::array::StringArray>()
1176 .ok_or_else(|| {
1177 AdbcError::with_message_and_status(
1178 "Expected string column",
1179 AdbcStatus::Internal,
1180 )
1181 })?;
1182 for i in 0..batch.num_rows() {
1183 let schema_name = schema_col.value(i).to_string();
1184 let tbl_name = name_col.value(i).to_string();
1185 let tbl_type = type_col.value(i).to_string();
1186 result
1187 .entry(schema_name)
1188 .or_default()
1189 .push((tbl_name, tbl_type));
1190 }
1191 }
1192 result
1193 } else {
1194 std::collections::HashMap::new()
1195 };
1196
1197 let columns: ColumnMetadataMap = if include_columns {
1199 let mut sql = "SELECT COLUMN_NAME, COLUMN_ORDINAL_POSITION, COLUMN_TYPE, COLUMN_SCHEMA, COLUMN_TABLE FROM SYS.EXA_ALL_COLUMNS".to_string();
1200 let mut conditions = Vec::new();
1201 if let Some(schema_filter) = db_schema {
1202 conditions.push(format!(
1203 "COLUMN_SCHEMA LIKE '{}'",
1204 schema_filter.replace('\'', "''")
1205 ));
1206 }
1207 if let Some(tbl_name) = table_name {
1208 conditions.push(format!(
1209 "COLUMN_TABLE LIKE '{}'",
1210 tbl_name.replace('\'', "''")
1211 ));
1212 }
1213 if let Some(col_name) = column_name {
1214 conditions.push(format!(
1215 "COLUMN_NAME LIKE '{}'",
1216 col_name.replace('\'', "''")
1217 ));
1218 }
1219 if !conditions.is_empty() {
1220 sql.push_str(" WHERE ");
1221 sql.push_str(&conditions.join(" AND "));
1222 }
1223 sql.push_str(" ORDER BY COLUMN_SCHEMA, COLUMN_TABLE, COLUMN_ORDINAL_POSITION");
1224
1225 let batches = {
1226 let conn_arc = Arc::clone(&conn_arc);
1227 get_runtime()
1228 .block_on(async {
1229 let mut conn = conn_arc.lock().await;
1230 conn.query(&sql).await
1231 })
1232 .map_err(to_adbc_error)?
1233 };
1234
1235 let mut result: ColumnMetadataMap = ColumnMetadataMap::new();
1236 for batch in &batches {
1237 let col_name_arr = batch
1238 .column(0)
1239 .as_any()
1240 .downcast_ref::<arrow::array::StringArray>()
1241 .unwrap();
1242 let col_ordinal = batch.column(1);
1243 let col_type_arr = batch
1244 .column(2)
1245 .as_any()
1246 .downcast_ref::<arrow::array::StringArray>()
1247 .unwrap();
1248 let col_schema_arr = batch
1249 .column(3)
1250 .as_any()
1251 .downcast_ref::<arrow::array::StringArray>()
1252 .unwrap();
1253 let col_table_arr = batch
1254 .column(4)
1255 .as_any()
1256 .downcast_ref::<arrow::array::StringArray>()
1257 .unwrap();
1258
1259 for i in 0..batch.num_rows() {
1260 let schema_name = col_schema_arr.value(i).to_string();
1261 let tbl_name_val = col_table_arr.value(i).to_string();
1262 let name = col_name_arr.value(i).to_string();
1263 let ordinal = get_int_value(col_ordinal.as_ref(), i).unwrap_or(i as i32 + 1);
1265 let type_name = col_type_arr.value(i).to_string();
1266 result
1267 .entry((schema_name, tbl_name_val))
1268 .or_default()
1269 .push((name, ordinal, type_name));
1270 }
1271 }
1272 result
1273 } else {
1274 std::collections::HashMap::new()
1275 };
1276
1277 let get_objects_schema = build_get_objects_schema();
1279
1280 let column_fields = Fields::from(vec![
1282 Field::new("column_name", DataType::Utf8, false),
1283 Field::new("ordinal_position", DataType::Int32, false),
1284 Field::new("xdbc_type_name", DataType::Utf8, true),
1285 ]);
1286
1287 let constraint_fields = Fields::from(vec![
1289 Field::new("constraint_name", DataType::Utf8, true),
1290 Field::new("constraint_type", DataType::Utf8, false),
1291 ]);
1292
1293 let table_fields = Fields::from(vec![
1295 Field::new("table_name", DataType::Utf8, false),
1296 Field::new("table_type", DataType::Utf8, false),
1297 Field::new(
1298 "table_columns",
1299 DataType::List(Arc::new(Field::new(
1300 "item",
1301 DataType::Struct(column_fields.clone()),
1302 true,
1303 ))),
1304 true,
1305 ),
1306 Field::new(
1307 "table_constraints",
1308 DataType::List(Arc::new(Field::new(
1309 "item",
1310 DataType::Struct(constraint_fields.clone()),
1311 true,
1312 ))),
1313 true,
1314 ),
1315 ]);
1316
1317 let schema_struct_fields = Fields::from(vec![
1319 Field::new("db_schema_name", DataType::Utf8, true),
1320 Field::new(
1321 "db_schema_tables",
1322 DataType::List(Arc::new(Field::new(
1323 "item",
1324 DataType::Struct(table_fields.clone()),
1325 true,
1326 ))),
1327 true,
1328 ),
1329 ]);
1330
1331 let mut catalog_name_builder = StringBuilder::new();
1333
1334 let schema_struct_builder = StructBuilder::from_fields(schema_struct_fields.clone(), 0);
1336 let mut schema_list_builder = ListBuilder::new(schema_struct_builder);
1337
1338 catalog_name_builder.append_value("EXA");
1339
1340 let builder_err = |field: &str| {
1341 AdbcError::with_message_and_status(
1342 format!("Failed to access Arrow builder for field '{field}'"),
1343 AdbcStatus::Internal,
1344 )
1345 };
1346
1347 if include_schemas {
1348 for schema_name in &schemas {
1349 let struct_builder = schema_list_builder.values();
1351 let schema_name_builder = struct_builder
1352 .field_builder::<StringBuilder>(0)
1353 .ok_or_else(|| builder_err("db_schema_name"))?;
1354 schema_name_builder.append_value(schema_name);
1355
1356 let tables_list_builder = struct_builder
1358 .field_builder::<ListBuilder<Box<dyn ArrayBuilder>>>(1)
1359 .ok_or_else(|| builder_err("db_schema_tables"))?;
1360
1361 if include_tables {
1362 if let Some(schema_tables) = tables.get(schema_name) {
1363 for (tbl_name, tbl_type) in schema_tables {
1364 let table_struct = tables_list_builder
1365 .values()
1366 .as_any_mut()
1367 .downcast_mut::<StructBuilder>()
1368 .ok_or_else(|| builder_err("db_schema_tables downcast"))?;
1369
1370 table_struct
1372 .field_builder::<StringBuilder>(0)
1373 .ok_or_else(|| builder_err("table_name"))?
1374 .append_value(tbl_name);
1375 table_struct
1377 .field_builder::<StringBuilder>(1)
1378 .ok_or_else(|| builder_err("table_type"))?
1379 .append_value(tbl_type);
1380
1381 let columns_list = table_struct
1383 .field_builder::<ListBuilder<Box<dyn ArrayBuilder>>>(2)
1384 .ok_or_else(|| builder_err("table_columns"))?;
1385 if include_columns {
1386 let key = (schema_name.clone(), tbl_name.clone());
1387 if let Some(cols) = columns.get(&key) {
1388 for (col_name, ordinal, type_name) in cols {
1389 let col_struct = columns_list
1390 .values()
1391 .as_any_mut()
1392 .downcast_mut::<StructBuilder>()
1393 .ok_or_else(|| builder_err("table_columns downcast"))?;
1394 col_struct
1395 .field_builder::<StringBuilder>(0)
1396 .ok_or_else(|| builder_err("column_name"))?
1397 .append_value(col_name);
1398 col_struct
1399 .field_builder::<Int32Builder>(1)
1400 .ok_or_else(|| builder_err("ordinal_position"))?
1401 .append_value(*ordinal);
1402 col_struct
1403 .field_builder::<StringBuilder>(2)
1404 .ok_or_else(|| builder_err("xdbc_type_name"))?
1405 .append_value(type_name);
1406 col_struct.append(true);
1407 }
1408 }
1409 columns_list.append(true);
1410 } else {
1411 columns_list.append(false); }
1413
1414 let constraints_list = table_struct
1416 .field_builder::<ListBuilder<Box<dyn ArrayBuilder>>>(3)
1417 .ok_or_else(|| builder_err("table_constraints"))?;
1418 constraints_list.append(false); table_struct.append(true);
1421 }
1422 }
1423 tables_list_builder.append(true);
1424 } else {
1425 tables_list_builder.append(false); }
1427
1428 struct_builder.append(true);
1429 }
1430 schema_list_builder.append(true);
1431 } else {
1432 schema_list_builder.append(false); }
1434
1435 let schema = Arc::new(get_objects_schema);
1436 let batch = RecordBatch::try_new(
1437 Arc::clone(&schema),
1438 vec![
1439 Arc::new(catalog_name_builder.finish()),
1440 Arc::new(schema_list_builder.finish()),
1441 ],
1442 )
1443 .map_err(|e| AdbcError::with_message_and_status(e.to_string(), AdbcStatus::Internal))?;
1444
1445 Ok(Box::new(VecRecordBatchReader::new(schema, vec![batch])))
1446 }
1447
1448 fn get_table_schema(
1449 &self,
1450 _catalog: Option<&str>,
1451 db_schema: Option<&str>,
1452 table_name: &str,
1453 ) -> AdbcResult<Schema> {
1454 use arrow::datatypes::Field;
1455
1456 let conn_arc = self.inner.as_ref().ok_or_else(|| {
1457 AdbcError::with_message_and_status(
1458 "Connection not established. Call new_statement() or connect first.",
1459 AdbcStatus::InvalidState,
1460 )
1461 })?;
1462 let conn_arc = Arc::clone(conn_arc);
1463
1464 let mut sql = format!(
1465 "SELECT COLUMN_NAME, COLUMN_TYPE, COLUMN_MAXSIZE, COLUMN_NUM_PREC, COLUMN_NUM_SCALE, COLUMN_IS_NULLABLE FROM SYS.EXA_ALL_COLUMNS WHERE COLUMN_TABLE = '{}'",
1466 table_name.replace('\'', "''")
1467 );
1468 if let Some(schema) = db_schema {
1469 sql.push_str(&format!(
1470 " AND COLUMN_SCHEMA = '{}'",
1471 schema.replace('\'', "''")
1472 ));
1473 }
1474 sql.push_str(" ORDER BY COLUMN_ORDINAL_POSITION");
1475
1476 let batches = get_runtime()
1477 .block_on(async {
1478 let mut conn = conn_arc.lock().await;
1479 conn.query(&sql).await
1480 })
1481 .map_err(to_adbc_error)?;
1482
1483 let mut fields = Vec::new();
1484 let mut found_any = false;
1485
1486 for batch in &batches {
1487 let col_name_arr = batch
1488 .column(0)
1489 .as_any()
1490 .downcast_ref::<arrow::array::StringArray>()
1491 .ok_or_else(|| {
1492 AdbcError::with_message_and_status(
1493 "Expected string column for COLUMN_NAME",
1494 AdbcStatus::Internal,
1495 )
1496 })?;
1497 let col_type_arr = batch
1498 .column(1)
1499 .as_any()
1500 .downcast_ref::<arrow::array::StringArray>()
1501 .ok_or_else(|| {
1502 AdbcError::with_message_and_status(
1503 "Expected string column for COLUMN_TYPE",
1504 AdbcStatus::Internal,
1505 )
1506 })?;
1507 let col_nullable = batch.column(5);
1508
1509 for i in 0..batch.num_rows() {
1510 found_any = true;
1511 let name = col_name_arr.value(i).to_string();
1512 let type_str = col_type_arr.value(i).to_string();
1513
1514 let nullable = get_nullable_value(col_nullable.as_ref(), i);
1516
1517 let exasol_type = parse_exasol_type_string(&type_str)?;
1519 let arrow_type =
1520 TypeMapper::exasol_to_arrow(&exasol_type, nullable).map_err(|e| {
1521 AdbcError::with_message_and_status(e.to_string(), AdbcStatus::Internal)
1522 })?;
1523
1524 fields.push(Field::new(name, arrow_type, nullable));
1525 }
1526 }
1527
1528 if !found_any {
1529 return Err(AdbcError::with_message_and_status(
1530 format!("Table '{}' not found", table_name),
1531 AdbcStatus::NotFound,
1532 ));
1533 }
1534
1535 Ok(Schema::new(fields))
1536 }
1537
1538 fn get_table_types(&self) -> AdbcResult<Box<dyn RecordBatchReader + Send>> {
1539 use arrow::array::builder::StringBuilder;
1540 use arrow::datatypes::{DataType, Field};
1541
1542 let schema = Arc::new(Schema::new(vec![Field::new(
1543 "table_type",
1544 DataType::Utf8,
1545 false,
1546 )]));
1547
1548 let mut builder = StringBuilder::new();
1549 builder.append_value("TABLE");
1550 builder.append_value("VIEW");
1551 builder.append_value("SYSTEM TABLE");
1552
1553 let batch = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(builder.finish())])
1554 .map_err(|e| AdbcError::with_message_and_status(e.to_string(), AdbcStatus::Internal))?;
1555
1556 Ok(Box::new(VecRecordBatchReader::new(schema, vec![batch])))
1557 }
1558
1559 fn get_statistic_names(&self) -> AdbcResult<Box<dyn RecordBatchReader + Send>> {
1560 use arrow::datatypes::{DataType, Field};
1561 let schema = Arc::new(Schema::new(vec![
1562 Field::new("statistic_name", DataType::Utf8, false),
1563 Field::new("statistic_key", DataType::Int16, false),
1564 ]));
1565 Ok(Box::new(VecRecordBatchReader::empty(schema)))
1566 }
1567
1568 fn get_statistics(
1569 &self,
1570 _catalog: Option<&str>,
1571 _db_schema: Option<&str>,
1572 _table_name: Option<&str>,
1573 _approximate: bool,
1574 ) -> AdbcResult<Box<dyn RecordBatchReader + Send>> {
1575 Err::<Box<dyn RecordBatchReader + Send>, _>(AdbcError::with_message_and_status(
1576 "get_statistics not yet implemented",
1577 AdbcStatus::NotImplemented,
1578 ))
1579 }
1580
1581 fn commit(&mut self) -> AdbcResult<()> {
1582 if self.auto_commit {
1583 return Err(AdbcError::with_message_and_status(
1584 "Cannot commit when autocommit is enabled",
1585 AdbcStatus::InvalidState,
1586 ));
1587 }
1588 let conn_arc = self.ensure_connected()?;
1589 get_runtime()
1590 .block_on(async {
1591 let mut conn = conn_arc.lock().await;
1592 conn.commit().await
1593 })
1594 .map_err(to_adbc_error)
1595 }
1596
1597 fn rollback(&mut self) -> AdbcResult<()> {
1598 if self.auto_commit {
1599 return Err(AdbcError::with_message_and_status(
1600 "Cannot rollback when autocommit is enabled",
1601 AdbcStatus::InvalidState,
1602 ));
1603 }
1604 let conn_arc = self.ensure_connected()?;
1605 get_runtime()
1606 .block_on(async {
1607 let mut conn = conn_arc.lock().await;
1608 conn.rollback().await
1609 })
1610 .map_err(to_adbc_error)
1611 }
1612
1613 fn read_partition(
1614 &self,
1615 _partition: impl AsRef<[u8]>,
1616 ) -> AdbcResult<Box<dyn RecordBatchReader + Send>> {
1617 Err::<Box<dyn RecordBatchReader + Send>, _>(AdbcError::with_message_and_status(
1618 "Partitioned results not supported",
1619 AdbcStatus::NotImplemented,
1620 ))
1621 }
1622}
1623
1624fn arrow_value_to_parameter(array: &dyn Array, row: usize) -> AdbcResult<Parameter> {
1635 use arrow::array::{
1636 BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array, Float64Array,
1637 Int16Array, Int32Array, Int64Array, LargeBinaryArray, LargeStringArray, StringArray,
1638 TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
1639 TimestampSecondArray,
1640 };
1641 use arrow::datatypes::DataType;
1642
1643 if array.is_null(row) {
1644 return Ok(Parameter::Null);
1645 }
1646
1647 let dt = array.data_type();
1648 match dt {
1649 DataType::Boolean => {
1650 let arr = array
1651 .as_any()
1652 .downcast_ref::<BooleanArray>()
1653 .expect("boolean downcast");
1654 Ok(Parameter::Boolean(arr.value(row)))
1655 }
1656 DataType::Int16 => {
1657 let arr = array
1658 .as_any()
1659 .downcast_ref::<Int16Array>()
1660 .expect("int16 downcast");
1661 Ok(Parameter::Integer(arr.value(row) as i64))
1662 }
1663 DataType::Int32 => {
1664 let arr = array
1665 .as_any()
1666 .downcast_ref::<Int32Array>()
1667 .expect("int32 downcast");
1668 Ok(Parameter::Integer(arr.value(row) as i64))
1669 }
1670 DataType::Int64 => {
1671 let arr = array
1672 .as_any()
1673 .downcast_ref::<Int64Array>()
1674 .expect("int64 downcast");
1675 Ok(Parameter::Integer(arr.value(row)))
1676 }
1677 DataType::Float32 => {
1678 let arr = array
1679 .as_any()
1680 .downcast_ref::<Float32Array>()
1681 .expect("float32 downcast");
1682 Ok(Parameter::Float(arr.value(row) as f64))
1683 }
1684 DataType::Float64 => {
1685 let arr = array
1686 .as_any()
1687 .downcast_ref::<Float64Array>()
1688 .expect("float64 downcast");
1689 Ok(Parameter::Float(arr.value(row)))
1690 }
1691 DataType::Utf8 => {
1692 let arr = array
1693 .as_any()
1694 .downcast_ref::<StringArray>()
1695 .expect("utf8 downcast");
1696 Ok(Parameter::String(arr.value(row).to_string()))
1697 }
1698 DataType::LargeUtf8 => {
1699 let arr = array
1700 .as_any()
1701 .downcast_ref::<LargeStringArray>()
1702 .expect("large utf8 downcast");
1703 Ok(Parameter::String(arr.value(row).to_string()))
1704 }
1705 DataType::Binary => {
1706 let arr = array
1707 .as_any()
1708 .downcast_ref::<BinaryArray>()
1709 .expect("binary downcast");
1710 Ok(Parameter::Binary(arr.value(row).to_vec()))
1711 }
1712 DataType::LargeBinary => {
1713 let arr = array
1714 .as_any()
1715 .downcast_ref::<LargeBinaryArray>()
1716 .expect("large binary downcast");
1717 Ok(Parameter::Binary(arr.value(row).to_vec()))
1718 }
1719 DataType::Date32 => {
1720 let arr = array
1722 .as_any()
1723 .downcast_ref::<Date32Array>()
1724 .expect("date32 downcast");
1725 let days = arr.value(row);
1726 let epoch = chrono::NaiveDate::from_ymd_opt(1970, 1, 1).unwrap();
1727 let date = epoch + chrono::Duration::days(days as i64);
1728 Ok(Parameter::String(date.format("%Y-%m-%d").to_string()))
1729 }
1730 DataType::Timestamp(unit, _tz) => {
1731 let nanos = match unit {
1733 arrow::datatypes::TimeUnit::Second => {
1734 let arr = array
1735 .as_any()
1736 .downcast_ref::<TimestampSecondArray>()
1737 .expect("timestamp_s downcast");
1738 arr.value(row) * 1_000_000_000
1739 }
1740 arrow::datatypes::TimeUnit::Millisecond => {
1741 let arr = array
1742 .as_any()
1743 .downcast_ref::<TimestampMillisecondArray>()
1744 .expect("timestamp_ms downcast");
1745 arr.value(row) * 1_000_000
1746 }
1747 arrow::datatypes::TimeUnit::Microsecond => {
1748 let arr = array
1749 .as_any()
1750 .downcast_ref::<TimestampMicrosecondArray>()
1751 .expect("timestamp_us downcast");
1752 arr.value(row) * 1_000
1753 }
1754 arrow::datatypes::TimeUnit::Nanosecond => {
1755 let arr = array
1756 .as_any()
1757 .downcast_ref::<TimestampNanosecondArray>()
1758 .expect("timestamp_ns downcast");
1759 arr.value(row)
1760 }
1761 };
1762 let secs = nanos.div_euclid(1_000_000_000);
1763 let subsec_nanos = nanos.rem_euclid(1_000_000_000) as u32;
1764 let dt = chrono::DateTime::from_timestamp(secs, subsec_nanos).ok_or_else(|| {
1765 AdbcError::with_message_and_status(
1766 format!("Invalid timestamp value: {nanos} nanos"),
1767 AdbcStatus::InvalidArguments,
1768 )
1769 })?;
1770 Ok(Parameter::String(
1771 dt.format("%Y-%m-%d %H:%M:%S%.6f").to_string(),
1772 ))
1773 }
1774 DataType::Decimal128(_precision, scale) => {
1775 let arr = array
1776 .as_any()
1777 .downcast_ref::<Decimal128Array>()
1778 .expect("decimal128 downcast");
1779 let raw = arr.value(row);
1780 if *scale <= 0 {
1782 Ok(Parameter::String(raw.to_string()))
1783 } else {
1784 let scale_u = *scale as u32;
1785 let divisor = 10_i128.pow(scale_u);
1786 let whole = raw / divisor;
1787 let frac = (raw % divisor).unsigned_abs();
1788 Ok(Parameter::String(format!(
1789 "{}{}.{:0>width$}",
1790 if raw < 0 && whole == 0 { "-" } else { "" },
1791 whole,
1792 frac,
1793 width = scale_u as usize
1794 )))
1795 }
1796 }
1797 other => Err(AdbcError::with_message_and_status(
1798 format!("Unsupported Arrow type for parameter binding: {other}"),
1799 AdbcStatus::NotImplemented,
1800 )),
1801 }
1802}
1803
1804pub struct FfiStatement {
1815 conn: Option<Arc<Mutex<ExaConnection>>>,
1817 uri: String,
1819 sql: Option<String>,
1821 bound_data: Option<RecordBatch>,
1823 options: std::collections::HashMap<String, OptionValue>,
1825 prepared: Option<PreparedStatement>,
1827 auto_commit: bool,
1829}
1830
1831impl FfiStatement {
1832 #[cfg(test)]
1833 fn new(uri: String) -> Self {
1834 Self {
1835 conn: None,
1836 uri,
1837 sql: None,
1838 bound_data: None,
1839 options: std::collections::HashMap::new(),
1840 prepared: None,
1841 auto_commit: true,
1842 }
1843 }
1844
1845 fn with_connection(conn: Option<Arc<Mutex<ExaConnection>>>, uri: String) -> Self {
1846 Self {
1847 conn,
1848 uri,
1849 sql: None,
1850 bound_data: None,
1851 options: std::collections::HashMap::new(),
1852 prepared: None,
1853 auto_commit: true,
1854 }
1855 }
1856
1857 fn execute_bulk_ingest(&mut self) -> AdbcResult<i64> {
1858 let target_table = self
1859 .options
1860 .get("adbc.ingest.target_table")
1861 .and_then(|v| match v {
1862 OptionValue::String(s) => Some(s.clone()),
1863 _ => None,
1864 })
1865 .ok_or_else(|| {
1866 AdbcError::with_message_and_status(
1867 "Target table not set for bulk ingestion",
1868 AdbcStatus::InvalidState,
1869 )
1870 })?;
1871
1872 if let Some(OptionValue::String(s)) = self.options.get("adbc.ingest.temporary") {
1873 if s == "true" {
1874 return Err(AdbcError::with_message_and_status(
1875 "Exasol does not support temporary tables",
1876 AdbcStatus::NotImplemented,
1877 ));
1878 }
1879 }
1880
1881 let target_schema =
1882 self.options
1883 .get("adbc.ingest.target_db_schema")
1884 .and_then(|v| match v {
1885 OptionValue::String(s) => Some(s.clone()),
1886 _ => None,
1887 });
1888
1889 let ingest_mode = self
1890 .options
1891 .get("adbc.ingest.mode")
1892 .and_then(|v| match v {
1893 OptionValue::String(s) => Some(s.clone()),
1894 _ => None,
1895 })
1896 .unwrap_or_else(|| "adbc.ingest.mode.append".to_string());
1897
1898 match ingest_mode.as_str() {
1899 "adbc.ingest.mode.create"
1900 | "adbc.ingest.mode.append"
1901 | "adbc.ingest.mode.create_append"
1902 | "adbc.ingest.mode.replace" => {}
1903 other => {
1904 return Err(AdbcError::with_message_and_status(
1905 format!("Unknown ingest mode: {}", other),
1906 AdbcStatus::InvalidArguments,
1907 ));
1908 }
1909 }
1910
1911 let qualified_name = build_qualified_table_name(target_schema.as_deref(), &target_table);
1912
1913 let batch = self.bound_data.take().ok_or_else(|| {
1914 AdbcError::with_message_and_status(
1915 "No data bound for bulk ingestion",
1916 AdbcStatus::InvalidState,
1917 )
1918 })?;
1919
1920 let conn_arc = self.conn.as_ref().ok_or_else(|| {
1921 AdbcError::with_message_and_status("No connection available", AdbcStatus::InvalidState)
1922 })?;
1923 let conn_arc = Arc::clone(conn_arc);
1924
1925 let arrow_schema = batch.schema();
1926
1927 match ingest_mode.as_str() {
1928 "adbc.ingest.mode.create" => {
1929 let ddl = generate_create_table_ddl(&qualified_name, &arrow_schema)?;
1930 let conn_arc2 = Arc::clone(&conn_arc);
1931 get_runtime()
1932 .block_on(async {
1933 let mut conn = conn_arc2.lock().await;
1934 conn.execute_update(&ddl).await
1935 })
1936 .map_err(to_adbc_error)?;
1937 }
1938 "adbc.ingest.mode.create_append" => {
1939 let ddl = generate_create_table_ddl(&qualified_name, &arrow_schema)?;
1940 let ddl = ddl.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS");
1941 let conn_arc2 = Arc::clone(&conn_arc);
1942 get_runtime()
1943 .block_on(async {
1944 let mut conn = conn_arc2.lock().await;
1945 conn.execute_update(&ddl).await
1946 })
1947 .map_err(to_adbc_error)?;
1948 }
1949 "adbc.ingest.mode.replace" => {
1950 let drop_ddl = format!("DROP TABLE IF EXISTS {}", qualified_name);
1951 let create_ddl = generate_create_table_ddl(&qualified_name, &arrow_schema)?;
1952 let conn_arc2 = Arc::clone(&conn_arc);
1953 get_runtime()
1954 .block_on(async {
1955 let mut conn = conn_arc2.lock().await;
1956 conn.execute_update(&drop_ddl).await?;
1957 conn.execute_update(&create_ddl).await
1958 })
1959 .map_err(to_adbc_error)?;
1960 }
1961 _ => {
1962 }
1964 }
1965
1966 let import_options = crate::import::arrow::ArrowImportOptions::default();
1967 let row_count = get_runtime()
1968 .block_on(async {
1969 let mut conn = conn_arc.lock().await;
1970 conn.import_from_record_batch(&qualified_name, &batch, import_options)
1971 .await
1972 })
1973 .map_err(to_adbc_error)?;
1974
1975 Ok(row_count as i64)
1976 }
1977}
1978
1979impl Optionable for FfiStatement {
1980 type Option = OptionStatement;
1981
1982 fn set_option(&mut self, key: Self::Option, value: OptionValue) -> AdbcResult<()> {
1983 match key {
1984 OptionStatement::Other(key) => {
1985 self.options.insert(key, value);
1986 }
1987 _ => {
1988 self.options.insert(key.as_ref().to_string(), value);
1990 }
1991 }
1992 Ok(())
1993 }
1994
1995 fn get_option_string(&self, key: Self::Option) -> AdbcResult<String> {
1996 let key_str = match key {
1997 OptionStatement::Other(ref k) => k.as_str(),
1998 _ => key.as_ref(),
1999 };
2000 if let Some(OptionValue::String(s)) = self.options.get(key_str) {
2001 Ok(s.clone())
2002 } else {
2003 Err(AdbcError::with_message_and_status(
2004 "Option not found or not a string",
2005 AdbcStatus::NotFound,
2006 ))
2007 }
2008 }
2009
2010 fn get_option_bytes(&self, key: Self::Option) -> AdbcResult<Vec<u8>> {
2011 let key_str = match key {
2012 OptionStatement::Other(ref k) => k.as_str(),
2013 _ => key.as_ref(),
2014 };
2015 if let Some(OptionValue::Bytes(b)) = self.options.get(key_str) {
2016 Ok(b.clone())
2017 } else {
2018 Err(AdbcError::with_message_and_status(
2019 "Option not found or not bytes",
2020 AdbcStatus::NotFound,
2021 ))
2022 }
2023 }
2024
2025 fn get_option_int(&self, key: Self::Option) -> AdbcResult<i64> {
2026 let key_str = match key {
2027 OptionStatement::Other(ref k) => k.as_str(),
2028 _ => key.as_ref(),
2029 };
2030 if let Some(OptionValue::Int(i)) = self.options.get(key_str) {
2031 Ok(*i)
2032 } else {
2033 Err(AdbcError::with_message_and_status(
2034 "Option not found or not an integer",
2035 AdbcStatus::NotFound,
2036 ))
2037 }
2038 }
2039
2040 fn get_option_double(&self, key: Self::Option) -> AdbcResult<f64> {
2041 let key_str = match key {
2042 OptionStatement::Other(ref k) => k.as_str(),
2043 _ => key.as_ref(),
2044 };
2045 if let Some(OptionValue::Double(d)) = self.options.get(key_str) {
2046 Ok(*d)
2047 } else {
2048 Err(AdbcError::with_message_and_status(
2049 "Option not found or not a double",
2050 AdbcStatus::NotFound,
2051 ))
2052 }
2053 }
2054}
2055
2056impl adbc_core::Statement for FfiStatement {
2057 fn bind(&mut self, batch: RecordBatch) -> AdbcResult<()> {
2058 self.bound_data = Some(batch);
2059 Ok(())
2060 }
2061
2062 fn bind_stream(&mut self, mut reader: Box<dyn RecordBatchReader + Send>) -> AdbcResult<()> {
2063 let mut batches = Vec::new();
2065 for batch_result in reader.by_ref() {
2066 let batch = batch_result.map_err(|e| {
2067 AdbcError::with_message_and_status(e.to_string(), AdbcStatus::Internal)
2068 })?;
2069 batches.push(batch);
2070 }
2071
2072 if batches.is_empty() {
2074 self.bound_data = None;
2075 } else if batches.len() == 1 {
2076 self.bound_data = Some(batches.remove(0));
2077 } else {
2078 let schema = batches[0].schema();
2080 let combined = concat_batches(&schema, &batches).map_err(|e| {
2081 AdbcError::with_message_and_status(e.to_string(), AdbcStatus::Internal)
2082 })?;
2083 self.bound_data = Some(combined);
2084 }
2085 Ok(())
2086 }
2087
2088 fn execute(&mut self) -> AdbcResult<Box<dyn RecordBatchReader + Send>> {
2089 if self.options.contains_key("adbc.ingest.target_table") {
2090 self.execute_bulk_ingest()?;
2091 return Ok(Box::new(VecRecordBatchReader::new(
2092 Arc::new(Schema::empty()),
2093 vec![],
2094 )));
2095 }
2096
2097 let sql = self.sql.as_ref().ok_or_else(|| {
2098 AdbcError::with_message_and_status("SQL query not set", AdbcStatus::InvalidState)
2099 })?;
2100
2101 if let Some(batch) = self.bound_data.take() {
2102 if self.prepared.is_none() {
2103 self.prepare()?;
2104 }
2105
2106 let prepared = self.prepared.as_mut().ok_or_else(|| {
2107 AdbcError::with_message_and_status(
2108 "Failed to prepare statement",
2109 AdbcStatus::InvalidState,
2110 )
2111 })?;
2112
2113 let conn_arc = self.conn.as_ref().ok_or_else(|| {
2114 AdbcError::with_message_and_status(
2115 "No connection available",
2116 AdbcStatus::InvalidState,
2117 )
2118 })?;
2119 let conn_arc = Arc::clone(conn_arc);
2120
2121 let num_rows = batch.num_rows();
2122 let mut all_batches: Vec<RecordBatch> = Vec::new();
2123
2124 for row_idx in 0..num_rows {
2125 for col_idx in 0..batch.num_columns() {
2126 let array = batch.column(col_idx);
2127 let param = arrow_value_to_parameter(array.as_ref(), row_idx)?;
2128 prepared.bind(col_idx, param).map_err(to_adbc_error)?;
2129 }
2130
2131 let batches = get_runtime()
2132 .block_on(async {
2133 let mut conn = conn_arc.lock().await;
2134 let result_set = conn.execute_prepared(prepared).await?;
2135 result_set.fetch_all().await
2136 })
2137 .map_err(to_adbc_error)?;
2138
2139 all_batches.extend(batches);
2140 prepared.clear_parameters();
2141 }
2142
2143 let schema = if all_batches.is_empty() {
2144 Arc::new(Schema::empty())
2145 } else {
2146 all_batches[0].schema()
2147 };
2148
2149 return Ok(Box::new(VecRecordBatchReader::new(schema, all_batches)));
2150 }
2151
2152 let sql = sql.clone();
2153
2154 let batches = if let Some(ref conn_arc) = self.conn {
2155 let conn_arc = Arc::clone(conn_arc);
2157 match get_runtime().block_on(async {
2158 let mut conn = conn_arc.lock().await;
2159 conn.query(&sql).await
2160 }) {
2161 Ok(batches) => batches,
2162 Err(QueryError::NoResultSet(_)) => vec![],
2163 Err(e) => return Err(to_adbc_error(e)),
2164 }
2165 } else {
2166 let uri = self.uri.clone();
2168 get_runtime()
2169 .block_on(async {
2170 let params: crate::connection::ConnectionParams = uri.parse()?;
2171 let mut conn = ExaConnection::from_params(params).await?;
2172 let result = conn.query(&sql).await;
2173 conn.close().await?;
2174 match result {
2175 Ok(batches) => Ok(batches),
2176 Err(QueryError::NoResultSet(_)) => Ok(vec![]),
2177 Err(e) => Err(ExasolError::from(e)),
2178 }
2179 })
2180 .map_err(to_adbc_error)?
2181 };
2182
2183 let schema = if batches.is_empty() {
2185 Arc::new(Schema::empty())
2186 } else {
2187 batches[0].schema()
2188 };
2189
2190 Ok(Box::new(VecRecordBatchReader::new(schema, batches)))
2191 }
2192
2193 fn execute_update(&mut self) -> AdbcResult<Option<i64>> {
2194 if self.options.contains_key("adbc.ingest.target_table") {
2195 let count = self.execute_bulk_ingest()?;
2196 return Ok(Some(count));
2197 }
2198
2199 let sql = self.sql.as_ref().ok_or_else(|| {
2200 AdbcError::with_message_and_status("SQL query not set", AdbcStatus::InvalidState)
2201 })?;
2202
2203 let sql = sql.clone();
2204
2205 if let Some(batch) = self.bound_data.take() {
2206 if self.prepared.is_none() {
2207 self.prepare()?;
2208 }
2209
2210 let prepared = self.prepared.as_mut().ok_or_else(|| {
2211 AdbcError::with_message_and_status(
2212 "Failed to prepare statement",
2213 AdbcStatus::InvalidState,
2214 )
2215 })?;
2216
2217 let conn_arc = self.conn.as_ref().ok_or_else(|| {
2218 AdbcError::with_message_and_status(
2219 "No connection available",
2220 AdbcStatus::InvalidState,
2221 )
2222 })?;
2223 let conn_arc = Arc::clone(conn_arc);
2224
2225 let num_rows = batch.num_rows();
2226 let mut total_count: i64 = 0;
2227
2228 for row_idx in 0..num_rows {
2229 for col_idx in 0..batch.num_columns() {
2230 let array = batch.column(col_idx);
2231 let param = arrow_value_to_parameter(array.as_ref(), row_idx)?;
2232 prepared.bind(col_idx, param).map_err(to_adbc_error)?;
2233 }
2234
2235 let count = get_runtime()
2236 .block_on(async {
2237 let mut conn = conn_arc.lock().await;
2238 conn.execute_prepared_update(prepared).await
2239 })
2240 .map_err(to_adbc_error)?;
2241
2242 total_count += count;
2243 prepared.clear_parameters();
2244 }
2245
2246 return Ok(Some(total_count));
2247 }
2248
2249 let count = if let Some(ref conn_arc) = self.conn {
2250 let conn_arc = Arc::clone(conn_arc);
2252 get_runtime()
2253 .block_on(async {
2254 let mut conn = conn_arc.lock().await;
2255 conn.execute_update(&sql).await
2256 })
2257 .map_err(to_adbc_error)?
2258 } else {
2259 let uri = self.uri.clone();
2261 get_runtime()
2262 .block_on(async {
2263 let params: crate::connection::ConnectionParams = uri.parse()?;
2264 let mut conn = ExaConnection::from_params(params).await?;
2265 let count = conn.execute_update(&sql).await?;
2266 conn.close().await?;
2267 Ok::<_, ExasolError>(count)
2268 })
2269 .map_err(to_adbc_error)?
2270 };
2271
2272 Ok(Some(count))
2273 }
2274
2275 fn execute_schema(&mut self) -> AdbcResult<Schema> {
2276 if self.prepared.is_none() {
2277 self.prepare()?;
2278 }
2279
2280 let sql = self.sql.as_ref().ok_or_else(|| {
2281 AdbcError::with_message_and_status("SQL query not set", AdbcStatus::InvalidState)
2282 })?;
2283
2284 let conn_arc = self.conn.as_ref().ok_or_else(|| {
2285 AdbcError::with_message_and_status("No connection available", AdbcStatus::InvalidState)
2286 })?;
2287 let conn_arc = Arc::clone(conn_arc);
2288 let sql = sql.clone();
2289 let batches = get_runtime()
2290 .block_on(async {
2291 let mut conn = conn_arc.lock().await;
2292 conn.query(&sql).await
2293 })
2294 .map_err(to_adbc_error)?;
2295
2296 if batches.is_empty() {
2297 Ok(Schema::empty())
2298 } else {
2299 Ok(batches[0].schema().as_ref().clone())
2300 }
2301 }
2302
2303 fn execute_partitions(&mut self) -> AdbcResult<PartitionedResult> {
2304 Err(AdbcError::with_message_and_status(
2305 "Partitioned execution not supported",
2306 AdbcStatus::NotImplemented,
2307 ))
2308 }
2309
2310 fn get_parameter_schema(&self) -> AdbcResult<Schema> {
2311 use arrow::datatypes::Field;
2312
2313 let prepared = self.prepared.as_ref().ok_or_else(|| {
2314 AdbcError::with_message_and_status(
2315 "Statement not prepared. Call prepare() first.",
2316 AdbcStatus::InvalidState,
2317 )
2318 })?;
2319
2320 let mut fields = Vec::new();
2321 let handle_ref = prepared.handle_ref();
2322 for (i, param_type) in handle_ref.parameter_types.iter().enumerate() {
2323 let exasol_type = transport_datatype_to_exasol(param_type)?;
2324 let arrow_type = TypeMapper::exasol_to_arrow(&exasol_type, true).map_err(|e| {
2325 AdbcError::with_message_and_status(e.to_string(), AdbcStatus::Internal)
2326 })?;
2327 let name = handle_ref
2328 .parameter_names
2329 .get(i)
2330 .and_then(|n| n.as_deref())
2331 .filter(|name| !name.is_empty())
2332 .map(String::from)
2333 .unwrap_or_else(|| format!("parameter_{}", i + 1));
2334 fields.push(Field::new(name, arrow_type, true));
2335 }
2336
2337 Ok(Schema::new(fields))
2338 }
2339
2340 fn prepare(&mut self) -> AdbcResult<()> {
2341 let sql = self.sql.as_ref().ok_or_else(|| {
2342 AdbcError::with_message_and_status("SQL query not set", AdbcStatus::InvalidState)
2343 })?;
2344 let sql = sql.clone();
2345
2346 let conn_arc = self.conn.as_ref().ok_or_else(|| {
2347 AdbcError::with_message_and_status("No connection available", AdbcStatus::InvalidState)
2348 })?;
2349 let conn_arc = Arc::clone(conn_arc);
2350
2351 let prepared_stmt = get_runtime()
2352 .block_on(async {
2353 let mut conn = conn_arc.lock().await;
2354 conn.prepare(&sql).await
2355 })
2356 .map_err(to_adbc_error)?;
2357
2358 self.prepared = Some(prepared_stmt);
2359
2360 Ok(())
2361 }
2362
2363 fn set_sql_query(&mut self, query: impl AsRef<str>) -> AdbcResult<()> {
2364 self.sql = Some(query.as_ref().to_string());
2365 Ok(())
2366 }
2367
2368 fn set_substrait_plan(&mut self, _plan: impl AsRef<[u8]>) -> AdbcResult<()> {
2369 Err(AdbcError::with_message_and_status(
2370 "Substrait plans not supported",
2371 AdbcStatus::NotImplemented,
2372 ))
2373 }
2374
2375 fn cancel(&mut self) -> AdbcResult<()> {
2376 Err(AdbcError::with_message_and_status(
2377 "Cancel not implemented",
2378 AdbcStatus::NotImplemented,
2379 ))
2380 }
2381}
2382
2383adbc_ffi::export_driver!(ExarrowDriverInit, FfiDriver);
2390
2391#[cfg(test)]
2392mod tests {
2393 use super::*;
2394 use adbc_core::{Driver, Statement};
2395 use arrow::datatypes::{DataType, Field};
2396
2397 #[test]
2398 fn test_ffi_driver_creation() {
2399 let mut driver = FfiDriver;
2400 let db = driver.new_database();
2401 assert!(db.is_ok());
2402 }
2403
2404 #[test]
2405 fn test_ffi_database_options() {
2406 let mut db = FfiDatabase::new();
2407
2408 db.set_option(
2410 OptionDatabase::Uri,
2411 "exasol://user:pass@localhost:8563".into(),
2412 )
2413 .unwrap();
2414
2415 let uri = db.get_option_string(OptionDatabase::Uri).unwrap();
2417 assert_eq!(uri, "exasol://user:pass@localhost:8563");
2418
2419 db.set_option(OptionDatabase::Username, "admin".into())
2421 .unwrap();
2422 let username = db.get_option_string(OptionDatabase::Username).unwrap();
2423 assert_eq!(username, "admin");
2424 }
2425
2426 #[test]
2427 fn test_ffi_database_build_uri_with_overrides() {
2428 let mut db = FfiDatabase::new();
2429
2430 db.set_option(
2432 OptionDatabase::Uri,
2433 "exasol://user:pass@localhost:8563/schema".into(),
2434 )
2435 .unwrap();
2436
2437 db.set_option(OptionDatabase::Username, "admin".into())
2439 .unwrap();
2440
2441 db.set_option(OptionDatabase::Password, "secret".into())
2443 .unwrap();
2444
2445 let uri = db.build_connection_uri().unwrap();
2446 assert!(uri.contains("admin"));
2447 assert!(uri.contains("secret"));
2448 assert!(uri.contains("localhost:8563/schema"));
2449 }
2450
2451 #[test]
2452 fn test_ffi_connection_options() {
2453 let mut conn = FfiConnection::new("exasol://user@localhost:8563".to_string());
2454
2455 conn.set_option(OptionConnection::AutoCommit, "false".into())
2457 .unwrap();
2458 let auto_commit = conn
2459 .get_option_string(OptionConnection::AutoCommit)
2460 .unwrap();
2461 assert_eq!(auto_commit, "false");
2462 }
2463
2464 #[test]
2465 fn test_ffi_statement_sql() {
2466 let mut stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2467
2468 stmt.set_sql_query("SELECT 1").unwrap();
2469 assert_eq!(stmt.sql, Some("SELECT 1".to_string()));
2470 }
2471
2472 #[test]
2473 fn test_generate_create_table_ddl_basic() {
2474 let schema = Schema::new(vec![
2475 Field::new("id", DataType::Int32, false),
2476 Field::new("name", DataType::Utf8, true),
2477 Field::new("score", DataType::Float64, true),
2478 ]);
2479
2480 let ddl = generate_create_table_ddl("test_table", &schema).unwrap();
2481 assert!(ddl.starts_with("CREATE TABLE test_table ("));
2482 assert!(ddl.contains("\"id\" DECIMAL(18,0) NOT NULL"));
2483 assert!(ddl.contains("\"name\" VARCHAR(2000000)"));
2484 assert!(ddl.contains("\"score\" DOUBLE"));
2485 }
2486
2487 #[test]
2488 fn test_generate_create_table_ddl_all_not_null() {
2489 let schema = Schema::new(vec![
2490 Field::new("a", DataType::Boolean, false),
2491 Field::new("b", DataType::Date32, false),
2492 ]);
2493
2494 let ddl = generate_create_table_ddl("my_table", &schema).unwrap();
2495 assert!(ddl.contains("\"a\" BOOLEAN NOT NULL"));
2496 assert!(ddl.contains("\"b\" DATE NOT NULL"));
2497 }
2498
2499 #[test]
2500 fn test_generate_create_table_ddl_empty_schema() {
2501 let schema = Schema::new(Vec::<Field>::new());
2502 let ddl = generate_create_table_ddl("empty_table", &schema).unwrap();
2503 assert_eq!(ddl, "CREATE TABLE empty_table ()");
2504 }
2505
2506 #[test]
2507 fn test_generate_create_table_ddl_column_name_escaping() {
2508 let schema = Schema::new(vec![Field::new(
2509 "col with \"quotes\"",
2510 DataType::Utf8,
2511 true,
2512 )]);
2513 let ddl = generate_create_table_ddl("t", &schema).unwrap();
2514 assert!(ddl.contains("\"col with \"\"quotes\"\"\""));
2515 }
2516
2517 #[test]
2518 fn test_parse_exasol_type_string_varchar() {
2519 let t = parse_exasol_type_string("VARCHAR(200)").unwrap();
2520 assert_eq!(t, ExasolType::Varchar { size: 200 });
2521 }
2522
2523 #[test]
2524 fn test_parse_exasol_type_string_decimal() {
2525 let t = parse_exasol_type_string("DECIMAL(10,3)").unwrap();
2526 assert_eq!(
2527 t,
2528 ExasolType::Decimal {
2529 precision: 10,
2530 scale: 3
2531 }
2532 );
2533 }
2534
2535 #[test]
2536 fn test_parse_exasol_type_string_boolean() {
2537 let t = parse_exasol_type_string("BOOLEAN").unwrap();
2538 assert_eq!(t, ExasolType::Boolean);
2539 }
2540
2541 #[test]
2542 fn test_parse_exasol_type_string_timestamp_with_tz() {
2543 let t = parse_exasol_type_string("TIMESTAMP WITH LOCAL TIME ZONE").unwrap();
2544 assert_eq!(
2545 t,
2546 ExasolType::Timestamp {
2547 with_local_time_zone: true
2548 }
2549 );
2550 }
2551
2552 #[test]
2553 fn test_parse_exasol_type_timestamp_with_precision() {
2554 for input in &["TIMESTAMP(3)", "TIMESTAMP(6)", "TIMESTAMP(9)", "TIMESTAMP"] {
2555 let t = parse_exasol_type_string(input).unwrap();
2556 assert_eq!(
2557 t,
2558 ExasolType::Timestamp {
2559 with_local_time_zone: false
2560 },
2561 "Failed for input: {}",
2562 input
2563 );
2564 }
2565 }
2566
2567 #[test]
2568 fn test_parse_exasol_type_timestamp_with_local_time_zone_precision() {
2569 for input in &[
2570 "TIMESTAMP WITH LOCAL TIME ZONE",
2571 "TIMESTAMP(3) WITH LOCAL TIME ZONE",
2572 "TIMESTAMP(6) WITH LOCAL TIME ZONE",
2573 ] {
2574 let t = parse_exasol_type_string(input).unwrap();
2575 assert_eq!(
2576 t,
2577 ExasolType::Timestamp {
2578 with_local_time_zone: true
2579 },
2580 "Failed for input: {}",
2581 input
2582 );
2583 }
2584 }
2585
2586 #[test]
2587 fn test_parse_exasol_type_string_unknown() {
2588 let result = parse_exasol_type_string("UNKNOWN_TYPE");
2589 assert!(result.is_err());
2590 }
2591
2592 #[test]
2593 fn test_transport_datatype_to_exasol_decimal() {
2594 let dt = TransportDataType {
2595 type_name: "DECIMAL".to_string(),
2596 precision: Some(18),
2597 scale: Some(0),
2598 size: None,
2599 character_set: None,
2600 with_local_time_zone: None,
2601 fraction: None,
2602 };
2603 let exasol = transport_datatype_to_exasol(&dt).unwrap();
2604 assert_eq!(
2605 exasol,
2606 ExasolType::Decimal {
2607 precision: 18,
2608 scale: 0
2609 }
2610 );
2611 }
2612
2613 #[test]
2614 fn test_transport_datatype_to_exasol_varchar() {
2615 let dt = TransportDataType {
2616 type_name: "VARCHAR".to_string(),
2617 precision: None,
2618 scale: None,
2619 size: Some(100),
2620 character_set: Some("UTF8".to_string()),
2621 with_local_time_zone: None,
2622 fraction: None,
2623 };
2624 let exasol = transport_datatype_to_exasol(&dt).unwrap();
2625 assert_eq!(exasol, ExasolType::Varchar { size: 100 });
2626 }
2627
2628 #[test]
2629 fn test_get_parameter_schema_not_prepared() {
2630 let stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2631 let result = stmt.get_parameter_schema();
2632 assert!(result.is_err());
2633 }
2634
2635 #[test]
2636 fn test_get_objects_invalid_catalog() {
2637 let conn = FfiConnection {
2638 uri: "exasol://user@localhost:8563".to_string(),
2639 inner: None,
2640 options: std::collections::HashMap::new(),
2641 auto_commit: true,
2642 current_schema: None,
2643 };
2644
2645 let result = adbc_core::Connection::get_objects(
2647 &conn,
2648 ObjectDepth::Catalogs,
2649 Some("OTHER"),
2650 None,
2651 None,
2652 None,
2653 None,
2654 );
2655 assert!(result.is_ok());
2656 }
2657
2658 #[test]
2659 fn test_get_objects_no_connection() {
2660 let conn = FfiConnection {
2661 uri: "exasol://user@localhost:8563".to_string(),
2662 inner: None,
2663 options: std::collections::HashMap::new(),
2664 auto_commit: true,
2665 current_schema: None,
2666 };
2667
2668 let result = adbc_core::Connection::get_objects(
2670 &conn,
2671 ObjectDepth::Schemas,
2672 Some("EXA"),
2673 None,
2674 None,
2675 None,
2676 None,
2677 );
2678 assert!(result.is_err());
2679 }
2680
2681 #[test]
2682 fn test_get_table_schema_no_connection() {
2683 let conn = FfiConnection {
2684 uri: "exasol://user@localhost:8563".to_string(),
2685 inner: None,
2686 options: std::collections::HashMap::new(),
2687 auto_commit: true,
2688 current_schema: None,
2689 };
2690
2691 let result = adbc_core::Connection::get_table_schema(&conn, None, None, "SOME_TABLE");
2692 assert!(result.is_err());
2693 }
2694
2695 #[test]
2696 fn test_build_get_objects_schema() {
2697 let schema = build_get_objects_schema();
2698 assert_eq!(schema.fields().len(), 2);
2699 assert_eq!(schema.field(0).name(), "catalog_name");
2700 assert_eq!(schema.field(1).name(), "catalog_db_schemas");
2701 }
2702
2703 #[test]
2704 fn test_bulk_ingest_temporary_not_supported() {
2705 let mut stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2706 stmt.set_option(
2707 OptionStatement::Other("adbc.ingest.target_table".to_string()),
2708 OptionValue::String("test_table".to_string()),
2709 )
2710 .unwrap();
2711 stmt.set_option(
2712 OptionStatement::Other("adbc.ingest.temporary".to_string()),
2713 OptionValue::String("true".to_string()),
2714 )
2715 .unwrap();
2716
2717 let batch = RecordBatch::try_new(
2718 Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])),
2719 vec![Arc::new(arrow::array::Int32Array::from(vec![1]))],
2720 )
2721 .unwrap();
2722 stmt.bind(batch).unwrap();
2723
2724 let result = stmt.execute_update();
2725 assert!(result.is_err());
2726 let err = result.unwrap_err();
2727 assert_eq!(err.status, AdbcStatus::NotImplemented);
2728 assert!(err.message.contains("temporary tables"));
2729 }
2730
2731 #[test]
2732 fn test_bulk_ingest_no_bound_data() {
2733 let mut stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2734 stmt.set_option(
2735 OptionStatement::Other("adbc.ingest.target_table".to_string()),
2736 OptionValue::String("test_table".to_string()),
2737 )
2738 .unwrap();
2739
2740 let result = stmt.execute_update();
2741 assert!(result.is_err());
2742 let err = result.unwrap_err();
2743 assert!(err.message.contains("No data bound"));
2744 }
2745
2746 #[test]
2747 fn test_bulk_ingest_no_connection() {
2748 let mut stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2749 stmt.set_option(
2750 OptionStatement::Other("adbc.ingest.target_table".to_string()),
2751 OptionValue::String("test_table".to_string()),
2752 )
2753 .unwrap();
2754
2755 let batch = RecordBatch::try_new(
2756 Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])),
2757 vec![Arc::new(arrow::array::Int32Array::from(vec![1]))],
2758 )
2759 .unwrap();
2760 stmt.bind(batch).unwrap();
2761
2762 let result = stmt.execute_update();
2763 assert!(result.is_err());
2764 let err = result.unwrap_err();
2765 assert!(err.message.contains("No connection available"));
2766 }
2767
2768 #[test]
2769 fn test_bulk_ingest_unknown_mode() {
2770 let mut stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2771 stmt.set_option(
2772 OptionStatement::Other("adbc.ingest.target_table".to_string()),
2773 OptionValue::String("test_table".to_string()),
2774 )
2775 .unwrap();
2776 stmt.set_option(
2777 OptionStatement::Other("adbc.ingest.mode".to_string()),
2778 OptionValue::String("adbc.ingest.mode.unknown".to_string()),
2779 )
2780 .unwrap();
2781
2782 let batch = RecordBatch::try_new(
2783 Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])),
2784 vec![Arc::new(arrow::array::Int32Array::from(vec![1]))],
2785 )
2786 .unwrap();
2787 stmt.bind(batch).unwrap();
2788
2789 let result = stmt.execute_update();
2790 assert!(result.is_err());
2791 let err = result.unwrap_err();
2792 assert!(err.message.contains("Unknown ingest mode"));
2793 }
2794
2795 #[test]
2796 fn test_bulk_ingest_detection_in_execute() {
2797 let mut stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2798 stmt.set_option(
2799 OptionStatement::Other("adbc.ingest.target_table".to_string()),
2800 OptionValue::String("test_table".to_string()),
2801 )
2802 .unwrap();
2803
2804 let result = stmt.execute();
2806 match result {
2807 Ok(_) => panic!("Expected error for missing bound data"),
2808 Err(err) => assert!(
2809 err.message.contains("No data bound"),
2810 "Expected 'No data bound' error, got: {}",
2811 err.message
2812 ),
2813 }
2814 }
2815
2816 #[test]
2817 fn test_autocommit_default_true() {
2818 let conn = FfiConnection::new("exasol://user@localhost:8563".to_string());
2819 assert!(conn.auto_commit);
2820 }
2821
2822 #[test]
2823 fn test_autocommit_set_false() {
2824 let mut conn = FfiConnection::new("exasol://user@localhost:8563".to_string());
2825 conn.set_option(OptionConnection::AutoCommit, "false".into())
2826 .unwrap();
2827 assert!(!conn.auto_commit);
2828 }
2829
2830 #[test]
2831 fn test_statement_inherits_autocommit() {
2832 let mut stmt = FfiStatement::new("exasol://user@localhost:8563".to_string());
2833 assert!(stmt.auto_commit);
2834 stmt.auto_commit = false;
2835 assert!(!stmt.auto_commit);
2836 }
2837
2838 #[test]
2839 fn test_parse_interval_year_to_month_with_precision() {
2840 let t = parse_exasol_type_string("INTERVAL YEAR(2) TO MONTH").unwrap();
2841 assert_eq!(t, ExasolType::IntervalYearToMonth);
2842
2843 let t = parse_exasol_type_string("INTERVAL YEAR(5) TO MONTH").unwrap();
2844 assert_eq!(t, ExasolType::IntervalYearToMonth);
2845 }
2846
2847 #[test]
2848 fn test_parse_interval_year_to_month_bare() {
2849 let t = parse_exasol_type_string("INTERVAL YEAR TO MONTH").unwrap();
2850 assert_eq!(t, ExasolType::IntervalYearToMonth);
2851 }
2852
2853 #[test]
2854 fn test_parse_interval_day_to_second_with_precision() {
2855 let t = parse_exasol_type_string("INTERVAL DAY(2) TO SECOND(3)").unwrap();
2856 assert_eq!(t, ExasolType::IntervalDayToSecond { precision: 3 });
2857
2858 let t = parse_exasol_type_string("INTERVAL DAY(9) TO SECOND(9)").unwrap();
2859 assert_eq!(t, ExasolType::IntervalDayToSecond { precision: 9 });
2860 }
2861
2862 #[test]
2863 fn test_parse_interval_day_to_second_bare() {
2864 let t = parse_exasol_type_string("INTERVAL DAY TO SECOND").unwrap();
2865 assert_eq!(t, ExasolType::IntervalDayToSecond { precision: 3 });
2866
2867 let t = parse_exasol_type_string("INTERVAL DAY TO SECOND(6)").unwrap();
2868 assert_eq!(t, ExasolType::IntervalDayToSecond { precision: 6 });
2869 }
2870}