1use std::sync::Arc;
2
3use arrow::array::{
4 ArrayRef, BinaryBuilder, BooleanBuilder, Date32Builder, Decimal128Builder, Float64Builder,
5 IntervalMonthDayNanoBuilder, StringBuilder, TimestampMicrosecondBuilder,
6};
7use arrow::datatypes::{DataType as ArrowDataType, Field, IntervalUnit, Schema, TimeUnit};
8use arrow::record_batch::{RecordBatch, RecordBatchOptions};
9
10use crate::error::TransportError;
11use crate::types::conversion::parse_interval;
12
13use super::constants::{
14 INTEGER_DECIMAL_PRECISION, IS_VARCHAR, PARAMETER_DESCRIPTION, R_COLUMN_COUNT, R_EMPTY,
15 R_EXCEPTION, R_HANDLE, R_MORE_ROWS, R_RESULT_SET, R_ROW_COUNT, R_STILL_EXECUTING, R_WARNING,
16 SMALLINT_DECIMAL_PRECISION, T_BIGDECIMAL, T_BINARY, T_BOOLEAN, T_CHAR, T_DATE, T_DECIMAL,
17 T_DOUBLE, T_GEOMETRY, T_HASHTYPE, T_INTEGER, T_INTERVAL_DAY, T_INTERVAL_YEAR, T_REAL,
18 T_SMALLDECIMAL, T_SMALLINT, T_TIMESTAMP, T_TIMESTAMP_LOCAL_TZ, T_TIMESTAMP_UTC,
19};
20
21type BuiltColumn = (Field, ArrayRef);
23
24#[derive(Debug, Clone)]
26pub struct NativeColumnMeta {
27 pub name: String,
28 pub type_id: u32,
29 pub precision: Option<i32>,
30 pub scale: Option<i32>,
31 pub is_varchar: bool,
32 pub max_len: Option<i32>,
33 pub interval_fraction: Option<i32>,
35}
36
37#[derive(Debug)]
39pub enum NativeResponse {
40 ResultSet {
41 handle: i32,
42 columns: Vec<NativeColumnMeta>,
43 batch: Option<arrow::record_batch::RecordBatch>,
44 total_rows: i64,
45 rows_received: i64,
46 },
47 PreparedStatement {
51 handle: i32,
52 parameters: Vec<NativeColumnMeta>,
53 result_columns: Vec<NativeColumnMeta>,
54 },
55 RowCount(i64),
56 Empty,
57 StillExecuting,
58 MoreRows(Vec<u8>),
59 Exception {
60 message: String,
61 sql_state: String,
62 },
63}
64
65#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct NativeWarning {
68 pub message: String,
69 pub sql_state: String,
70}
71
72#[derive(Debug)]
74pub struct NativeResponseEnvelope {
75 pub warnings: Vec<NativeWarning>,
76 pub terminal: NativeResponse,
77}
78
79pub fn parse_response(data: &[u8]) -> Result<NativeResponseEnvelope, TransportError> {
81 if data.is_empty() {
82 return Ok(NativeResponseEnvelope {
83 warnings: Vec::new(),
84 terminal: NativeResponse::Empty,
85 });
86 }
87
88 if data.len() == 4 && i32::from_le_bytes([data[0], data[1], data[2], data[3]]) == 0 {
89 return Ok(NativeResponseEnvelope {
90 warnings: Vec::new(),
91 terminal: NativeResponse::Empty,
92 });
93 }
94
95 if let Some(envelope) = try_parse_counted_envelope(data)? {
96 return Ok(envelope);
97 }
98
99 parse_legacy_response(data)
100}
101
102fn try_parse_counted_envelope(
103 data: &[u8],
104) -> Result<Option<NativeResponseEnvelope>, TransportError> {
105 if data.len() < 5 {
106 return Ok(None);
107 }
108
109 let mut probe_offset = 0;
110 let result_count = read_i32(data, &mut probe_offset)?;
111 if !(1..=64).contains(&result_count) {
112 return Ok(None);
113 }
114
115 let first_part_type = data[probe_offset] as i8;
116 if !is_known_response_type(first_part_type) {
117 return Ok(None);
118 }
119
120 let mut offset = 0;
121 let result_count = read_i32(data, &mut offset)?;
122
123 let mut warnings = Vec::new();
124 let mut terminal = None;
125
126 for part_idx in 0..(result_count as usize) {
127 let result_type = read_u8(data, &mut offset)? as i8;
128 match result_type {
129 R_WARNING => warnings.push(parse_warning(data, &mut offset)?),
130 R_COLUMN_COUNT => {
131 let _ = read_i32(data, &mut offset)?;
132 }
133 R_RESULT_SET => {
134 let response = parse_result_set_at(data, &mut offset)?;
135 assign_terminal_result(&mut terminal, response)?;
136 }
137 R_HANDLE => {
138 let handle_envelope = parse_handle_only_at(data, &mut offset)?;
139 warnings.extend(handle_envelope.warnings);
140 assign_terminal_result(&mut terminal, handle_envelope.terminal)?;
141 }
142 R_ROW_COUNT => {
143 let response = parse_row_count_at(data, &mut offset)?;
144 assign_terminal_result(&mut terminal, response)?;
145 }
146 R_EXCEPTION => {
147 let response = parse_exception_at(data, &mut offset)?;
148 assign_terminal_result(&mut terminal, response)?;
149 }
150 R_EMPTY => {
151 assign_terminal_result(&mut terminal, NativeResponse::Empty)?;
152 }
153 R_STILL_EXECUTING => {
154 assign_terminal_result(&mut terminal, NativeResponse::StillExecuting)?;
155 }
156 R_MORE_ROWS => {
157 if part_idx + 1 != result_count as usize {
158 return Err(TransportError::ProtocolError(
159 "MoreRows must be the final native response part".into(),
160 ));
161 }
162 let response = NativeResponse::MoreRows(data[offset..].to_vec());
163 assign_terminal_result(&mut terminal, response)?;
164 offset = data.len();
165 }
166 _ => {
167 let preview_end = data.len().min(64);
168 return Err(TransportError::ProtocolError(format!(
169 "Unknown response type: {} at offset {} (data {:02x?})",
170 result_type,
171 offset.saturating_sub(1),
172 &data[..preview_end]
173 )));
174 }
175 }
176 }
177
178 if offset != data.len() {
179 return Ok(None);
180 }
181
182 Ok(Some(NativeResponseEnvelope {
183 warnings,
184 terminal: terminal.unwrap_or(NativeResponse::Empty),
185 }))
186}
187
188fn parse_legacy_response(data: &[u8]) -> Result<NativeResponseEnvelope, TransportError> {
189 let mut offset = 0;
190 let response = parse_legacy_response_at(data, &mut offset)?;
191 if offset != data.len() {
192 return Err(TransportError::ProtocolError(format!(
193 "Trailing bytes after legacy response: remaining {:02x?}",
194 &data[offset..]
195 )));
196 }
197 Ok(response)
198}
199
200fn parse_legacy_response_at(
201 data: &[u8],
202 offset: &mut usize,
203) -> Result<NativeResponseEnvelope, TransportError> {
204 let result_type = read_u8(data, offset)? as i8;
205 parse_legacy_response_body(data, offset, result_type)
206}
207
208fn parse_legacy_response_body(
209 data: &[u8],
210 offset: &mut usize,
211 result_type: i8,
212) -> Result<NativeResponseEnvelope, TransportError> {
213 match result_type {
214 R_RESULT_SET => Ok(NativeResponseEnvelope {
215 warnings: Vec::new(),
216 terminal: parse_result_set_at(data, offset)?,
217 }),
218 R_HANDLE => parse_handle_only_at(data, offset),
219 R_ROW_COUNT => Ok(NativeResponseEnvelope {
220 warnings: Vec::new(),
221 terminal: parse_row_count_at(data, offset)?,
222 }),
223 R_EXCEPTION => Ok(NativeResponseEnvelope {
224 warnings: Vec::new(),
225 terminal: parse_exception_at(data, offset)?,
226 }),
227 R_WARNING => Ok(NativeResponseEnvelope {
228 warnings: { vec![parse_warning(data, offset)?] },
229 terminal: NativeResponse::Empty,
230 }),
231 R_COLUMN_COUNT => {
232 let _ = read_i32(data, offset)?;
233 Ok(NativeResponseEnvelope {
234 warnings: Vec::new(),
235 terminal: NativeResponse::Empty,
236 })
237 }
238 R_EMPTY => Ok(NativeResponseEnvelope {
239 warnings: Vec::new(),
240 terminal: NativeResponse::Empty,
241 }),
242 R_STILL_EXECUTING => Ok(NativeResponseEnvelope {
243 warnings: Vec::new(),
244 terminal: {
245 consume_status_message(data, offset)?;
246 NativeResponse::StillExecuting
247 },
248 }),
249 R_MORE_ROWS => {
250 let remaining = data[*offset..].to_vec();
251 *offset = data.len();
252 Ok(NativeResponseEnvelope {
253 warnings: Vec::new(),
254 terminal: NativeResponse::MoreRows(remaining),
255 })
256 }
257 _ => Err(TransportError::ProtocolError(format!(
258 "Unknown response type: {}",
259 result_type
260 ))),
261 }
262}
263
264fn is_known_response_type(result_type: i8) -> bool {
265 matches!(
266 result_type,
267 R_RESULT_SET
268 | R_HANDLE
269 | R_ROW_COUNT
270 | R_COLUMN_COUNT
271 | R_WARNING
272 | R_STILL_EXECUTING
273 | R_MORE_ROWS
274 | R_EXCEPTION
275 | R_EMPTY
276 )
277}
278
279fn consume_status_message(data: &[u8], offset: &mut usize) -> Result<(), TransportError> {
280 if *offset >= data.len() {
281 return Ok(());
282 }
283 let msg_len = read_i32(data, offset)? as usize;
284 if *offset + msg_len > data.len() {
285 return Err(TransportError::ProtocolError(
286 "StillExecuting status message truncated".into(),
287 ));
288 }
289 *offset += msg_len;
290 Ok(())
291}
292
293fn assign_terminal_result(
294 slot: &mut Option<NativeResponse>,
295 response: NativeResponse,
296) -> Result<(), TransportError> {
297 if slot.is_some() {
298 return Err(TransportError::ProtocolError(
299 "Multiple terminal results in native response envelope".into(),
300 ));
301 }
302 *slot = Some(response);
303 Ok(())
304}
305
306fn parse_handle_only_at(
316 data: &[u8],
317 offset: &mut usize,
318) -> Result<NativeResponseEnvelope, TransportError> {
319 let statement_handle = read_i32(data, offset)?;
320
321 let mut warnings = Vec::new();
322 let mut parameters = Vec::new();
323 let mut result_columns = Vec::new();
324 while *offset < data.len() {
325 let sub_response = parse_legacy_response_at(data, offset)?;
326 warnings.extend(sub_response.warnings);
327 match sub_response.terminal {
328 NativeResponse::ResultSet {
329 handle: sub_handle,
330 columns,
331 ..
332 } => {
333 if sub_handle == PARAMETER_DESCRIPTION {
334 parameters = columns;
335 } else {
336 result_columns = columns;
337 }
338 }
339 NativeResponse::Exception { message, sql_state } => {
340 return Err(TransportError::ProtocolError(format!(
341 "Exasol reported an exception in a prepared-statement reply sub-result: {message} (SQLSTATE {sql_state})"
342 )));
343 }
344 _ => {}
345 }
346 }
347
348 Ok(NativeResponseEnvelope {
349 warnings,
350 terminal: NativeResponse::PreparedStatement {
351 handle: statement_handle,
352 parameters,
353 result_columns,
354 },
355 })
356}
357
358#[cfg(test)]
359fn parse_result_set(data: &[u8]) -> Result<NativeResponse, TransportError> {
360 parse_result_set_at_end(data)
361}
362
363#[cfg(test)]
364fn parse_result_set_at_end(data: &[u8]) -> Result<NativeResponse, TransportError> {
365 let mut offset = 0;
366 let response = parse_result_set_at(data, &mut offset)?;
367 if offset != data.len() {
368 return Err(TransportError::ProtocolError(format!(
369 "Trailing bytes after result-set response: remaining {:02x?}",
370 &data[offset..]
371 )));
372 }
373 Ok(response)
374}
375
376fn parse_result_set_at(data: &[u8], offset: &mut usize) -> Result<NativeResponse, TransportError> {
377 let handle = read_i32(data, offset)?;
378 let num_columns = read_i32(data, offset)? as usize;
379 let total_rows = read_i64(data, offset)?;
380 let rows_received = read_i64(data, offset)?;
381 if rows_received < 0 {
382 return Err(TransportError::ProtocolError(format!(
383 "Invalid row count from server: {}",
384 rows_received
385 )));
386 }
387
388 let mut columns = Vec::with_capacity(num_columns);
389 for _ in 0..num_columns {
390 columns.push(parse_column_meta(data, offset)?);
391 }
392
393 let batch = if num_columns == 0 {
394 None
395 } else {
396 Some(build_batch_from_wire(
397 data,
398 offset,
399 &columns,
400 rows_received as usize,
401 )?)
402 };
403
404 Ok(NativeResponse::ResultSet {
405 handle,
406 columns,
407 batch,
408 total_rows,
409 rows_received,
410 })
411}
412
413fn parse_column_meta(data: &[u8], offset: &mut usize) -> Result<NativeColumnMeta, TransportError> {
414 let name_len = read_i32(data, offset)? as usize;
415 if *offset + name_len > data.len() {
416 return Err(TransportError::ProtocolError(
417 "Column name truncated".into(),
418 ));
419 }
420 let name = std::str::from_utf8(&data[*offset..*offset + name_len])
421 .map_err(|e| TransportError::ProtocolError(format!("Invalid column name UTF-8: {}", e)))?
422 .to_owned();
423 *offset += name_len;
424
425 let type_id = read_i32(data, offset)? as u32;
426
427 let mut precision = None;
428 let mut scale = None;
429 let mut is_varchar = false;
430 let mut max_len = None;
431 let mut interval_fraction = None;
432
433 match type_id {
434 T_DECIMAL | T_SMALLDECIMAL | T_BIGDECIMAL => {
435 precision = Some(read_i32(data, offset)?);
436 scale = Some(read_i32(data, offset)?);
437 }
438 T_CHAR | T_GEOMETRY | T_HASHTYPE => {
439 if *offset >= data.len() {
441 return Err(TransportError::ProtocolError(
442 "CHAR metadata truncated".into(),
443 ));
444 }
445 let vc_flag = data[*offset];
446 *offset += 1;
447 is_varchar = (vc_flag & IS_VARCHAR) != 0;
448 max_len = Some(read_i32(data, offset)?);
449 let _octet_len = read_i32(data, offset)?;
451 }
452 T_INTERVAL_YEAR => {
453 if *offset >= data.len() {
455 return Err(TransportError::ProtocolError(
456 "INTERVAL metadata truncated".into(),
457 ));
458 }
459 let vc_flag = data[*offset];
460 *offset += 1;
461 is_varchar = (vc_flag & IS_VARCHAR) != 0;
462 max_len = Some(read_i32(data, offset)?);
463 let _octet_len = read_i32(data, offset)?;
464 let _interval_precision = read_i32(data, offset)?;
466 }
467 T_INTERVAL_DAY => {
468 if *offset >= data.len() {
470 return Err(TransportError::ProtocolError(
471 "INTERVAL metadata truncated".into(),
472 ));
473 }
474 let vc_flag = data[*offset];
475 *offset += 1;
476 is_varchar = (vc_flag & IS_VARCHAR) != 0;
477 max_len = Some(read_i32(data, offset)?);
478 let _octet_len = read_i32(data, offset)?;
479 let _interval_precision = read_i32(data, offset)?;
481 interval_fraction = Some(read_i32(data, offset)?);
482 }
483 T_TIMESTAMP | T_TIMESTAMP_LOCAL_TZ | T_TIMESTAMP_UTC => {
484 let _ts_precision = read_i32(data, offset)?;
486 }
487 _ => {}
488 }
489
490 Ok(NativeColumnMeta {
491 name,
492 type_id,
493 precision,
494 scale,
495 is_varchar,
496 max_len,
497 interval_fraction,
498 })
499}
500
501fn parse_row_count_at(data: &[u8], offset: &mut usize) -> Result<NativeResponse, TransportError> {
502 let count = read_i64(data, offset)?;
503 Ok(NativeResponse::RowCount(count))
504}
505
506fn parse_exception_at(data: &[u8], offset: &mut usize) -> Result<NativeResponse, TransportError> {
507 let warning = parse_message_part(data, offset)?;
508 Ok(NativeResponse::Exception {
509 message: warning.message,
510 sql_state: warning.sql_state,
511 })
512}
513
514fn parse_warning(data: &[u8], offset: &mut usize) -> Result<NativeWarning, TransportError> {
515 parse_message_part(data, offset)
516}
517
518fn parse_message_part(data: &[u8], offset: &mut usize) -> Result<NativeWarning, TransportError> {
519 let msg_len = read_i32(data, offset)? as usize;
520 if *offset + msg_len > data.len() {
521 return Err(TransportError::ProtocolError(
522 "Exception message truncated".into(),
523 ));
524 }
525 let message = std::str::from_utf8(&data[*offset..*offset + msg_len])
526 .map_err(|e| TransportError::ProtocolError(format!("Invalid exception UTF-8: {}", e)))?
527 .to_owned();
528 *offset += msg_len;
529
530 let sql_state = if *offset + 5 <= data.len() {
531 std::str::from_utf8(&data[*offset..*offset + 5])
532 .unwrap_or("?????")
533 .to_owned()
534 } else {
535 "?????".to_owned()
536 };
537 if *offset + 5 <= data.len() {
538 *offset += 5;
539 }
540
541 Ok(NativeWarning { message, sql_state })
542}
543
544fn build_batch_from_wire(
548 data: &[u8],
549 offset: &mut usize,
550 column_metas: &[NativeColumnMeta],
551 num_rows: usize,
552) -> Result<RecordBatch, TransportError> {
553 if column_metas.is_empty() {
554 return RecordBatch::try_new_with_options(
555 Arc::new(Schema::empty()),
556 vec![],
557 &RecordBatchOptions::new().with_row_count(Some(num_rows)),
558 )
559 .map_err(|e| {
560 TransportError::ProtocolError(format!("Failed to create empty RecordBatch: {}", e))
561 });
562 }
563
564 let mut fields = Vec::with_capacity(column_metas.len());
565 let mut arrays: Vec<ArrayRef> = Vec::with_capacity(column_metas.len());
566
567 for meta in column_metas {
568 let (field, array) = fill_column_builder(data, offset, meta, num_rows)?;
569 fields.push(field);
570 arrays.push(array);
571 }
572
573 RecordBatch::try_new(Arc::new(Schema::new(fields)), arrays)
574 .map_err(|e| TransportError::ProtocolError(format!("Failed to create RecordBatch: {}", e)))
575}
576
577fn decimal_precision_scale(meta: &NativeColumnMeta, default_precision: i32) -> (u8, i8) {
581 let raw_precision = meta.precision.unwrap_or(default_precision);
582 let precision = raw_precision.clamp(1, 38) as u8;
583 let raw_scale = meta.scale.unwrap_or(0);
584 let scale = raw_scale.clamp(i8::MIN as i32, i8::MAX as i32) as i8;
585 (precision, scale)
586}
587
588fn fill_column_builder(
598 data: &[u8],
599 offset: &mut usize,
600 meta: &NativeColumnMeta,
601 num_rows: usize,
602) -> Result<BuiltColumn, TransportError> {
603 match meta.type_id {
604 T_DOUBLE => fill_double_column(data, offset, &meta.name, num_rows),
605 T_REAL => fill_real_column(data, offset, &meta.name, num_rows),
606 T_INTEGER => fill_decimal_column(
607 data,
608 offset,
609 &whole_number_meta(meta, INTEGER_DECIMAL_PRECISION),
610 num_rows,
611 ),
612 T_SMALLINT => fill_smalldecimal_column(
613 data,
614 offset,
615 &whole_number_meta(meta, SMALLINT_DECIMAL_PRECISION),
616 num_rows,
617 ),
618 T_BOOLEAN => fill_boolean_column(data, offset, &meta.name, num_rows),
619 T_BINARY | T_GEOMETRY | T_HASHTYPE => {
620 fill_binary_column(data, offset, &meta.name, num_rows)
621 }
622 T_INTERVAL_YEAR | T_INTERVAL_DAY => {
623 fill_interval_column(data, offset, &meta.name, num_rows)
624 }
625 T_SMALLDECIMAL => fill_smalldecimal_column(data, offset, meta, num_rows),
626 T_DECIMAL => fill_decimal_column(data, offset, meta, num_rows),
627 T_BIGDECIMAL => fill_bigdecimal_column(data, offset, meta, num_rows),
628 T_DATE => fill_date_column(data, offset, &meta.name, num_rows),
629 T_TIMESTAMP | T_TIMESTAMP_LOCAL_TZ | T_TIMESTAMP_UTC => {
630 fill_timestamp_column(data, offset, meta, num_rows)
631 }
632 _ => fill_string_builder(data, offset, &meta.name, num_rows),
633 }
634}
635
636fn whole_number_meta(meta: &NativeColumnMeta, precision: i32) -> NativeColumnMeta {
637 NativeColumnMeta {
638 precision: Some(precision),
639 scale: Some(0),
640 ..meta.clone()
641 }
642}
643
644fn fill_double_column(
645 data: &[u8],
646 offset: &mut usize,
647 name: &str,
648 num_rows: usize,
649) -> Result<BuiltColumn, TransportError> {
650 let mut builder = Float64Builder::with_capacity(num_rows);
651 for _ in 0..num_rows {
652 if read_u8(data, offset)? == 0 {
653 builder.append_null();
654 } else {
655 builder.append_value(read_f64(data, offset)?);
656 }
657 }
658 Ok((
659 Field::new(name, ArrowDataType::Float64, true),
660 Arc::new(builder.finish()),
661 ))
662}
663
664fn fill_real_column(
665 data: &[u8],
666 offset: &mut usize,
667 name: &str,
668 num_rows: usize,
669) -> Result<BuiltColumn, TransportError> {
670 let mut builder = Float64Builder::with_capacity(num_rows);
671 for _ in 0..num_rows {
672 if read_u8(data, offset)? == 0 {
673 builder.append_null();
674 } else {
675 builder.append_value(read_f32(data, offset)? as f64);
676 }
677 }
678 Ok((
679 Field::new(name, ArrowDataType::Float64, true),
680 Arc::new(builder.finish()),
681 ))
682}
683
684fn fill_boolean_column(
685 data: &[u8],
686 offset: &mut usize,
687 name: &str,
688 num_rows: usize,
689) -> Result<BuiltColumn, TransportError> {
690 let mut builder = BooleanBuilder::with_capacity(num_rows);
691 for _ in 0..num_rows {
692 if read_u8(data, offset)? == 0 {
693 builder.append_null();
694 } else {
695 builder.append_value(read_u8(data, offset)? != 0);
696 }
697 }
698 Ok((
699 Field::new(name, ArrowDataType::Boolean, true),
700 Arc::new(builder.finish()),
701 ))
702}
703
704fn fill_binary_column(
705 data: &[u8],
706 offset: &mut usize,
707 name: &str,
708 num_rows: usize,
709) -> Result<BuiltColumn, TransportError> {
710 let mut builder = BinaryBuilder::with_capacity(num_rows, num_rows * 16);
711 for _ in 0..num_rows {
712 if read_u8(data, offset)? == 0 {
713 builder.append_null();
714 } else {
715 builder.append_value(read_length_prefixed(data, offset, "Binary data truncated")?);
716 }
717 }
718 Ok((
719 Field::new(name, ArrowDataType::Binary, true),
720 Arc::new(builder.finish()),
721 ))
722}
723
724fn fill_interval_column(
725 data: &[u8],
726 offset: &mut usize,
727 name: &str,
728 num_rows: usize,
729) -> Result<BuiltColumn, TransportError> {
730 let mut builder = IntervalMonthDayNanoBuilder::with_capacity(num_rows);
731 for _ in 0..num_rows {
732 if read_u8(data, offset)? == 0 {
733 builder.append_null();
734 } else {
735 let bytes = read_length_prefixed(data, offset, "Interval data truncated")?;
736 let text = std::str::from_utf8(bytes).map_err(|e| {
737 TransportError::ProtocolError(format!(
738 "Invalid UTF-8 in INTERVAL column {name}: {e}"
739 ))
740 })?;
741 let interval = parse_interval(text).map_err(|reason| {
742 TransportError::ProtocolError(format!(
743 "Invalid INTERVAL value '{text}' in column {name}: {reason}"
744 ))
745 })?;
746 builder.append_value(interval);
747 }
748 }
749 Ok((
750 Field::new(
751 name,
752 ArrowDataType::Interval(IntervalUnit::MonthDayNano),
753 true,
754 ),
755 Arc::new(builder.finish()),
756 ))
757}
758
759fn fill_smalldecimal_column(
760 data: &[u8],
761 offset: &mut usize,
762 meta: &NativeColumnMeta,
763 num_rows: usize,
764) -> Result<BuiltColumn, TransportError> {
765 let data_type = decimal_data_type(meta, 9);
766 let mut builder = Decimal128Builder::with_capacity(num_rows).with_data_type(data_type.clone());
767 for _ in 0..num_rows {
768 if read_u8(data, offset)? == 0 {
769 builder.append_null();
770 } else {
771 builder.append_value(read_i32(data, offset)? as i128);
772 }
773 }
774 Ok((
775 Field::new(&meta.name, data_type, true),
776 Arc::new(builder.finish()),
777 ))
778}
779
780fn fill_decimal_column(
781 data: &[u8],
782 offset: &mut usize,
783 meta: &NativeColumnMeta,
784 num_rows: usize,
785) -> Result<BuiltColumn, TransportError> {
786 let data_type = decimal_data_type(meta, 18);
787 let mut builder = Decimal128Builder::with_capacity(num_rows).with_data_type(data_type.clone());
788 let use_i32 = meta.precision.unwrap_or(18) <= 9;
789 for _ in 0..num_rows {
790 if read_u8(data, offset)? == 0 {
791 builder.append_null();
792 } else if use_i32 {
793 builder.append_value(read_i32(data, offset)? as i128);
794 } else {
795 builder.append_value(read_i64(data, offset)? as i128);
796 }
797 }
798 Ok((
799 Field::new(&meta.name, data_type, true),
800 Arc::new(builder.finish()),
801 ))
802}
803
804fn fill_bigdecimal_column(
805 data: &[u8],
806 offset: &mut usize,
807 meta: &NativeColumnMeta,
808 num_rows: usize,
809) -> Result<BuiltColumn, TransportError> {
810 let data_type = decimal_data_type(meta, 36);
811 let mut builder = Decimal128Builder::with_capacity(num_rows).with_data_type(data_type.clone());
812 for _ in 0..num_rows {
813 if read_u8(data, offset)? == 0 {
814 builder.append_null();
815 } else {
816 builder.append_value(read_i128(data, offset)?);
817 }
818 }
819 Ok((
820 Field::new(&meta.name, data_type, true),
821 Arc::new(builder.finish()),
822 ))
823}
824
825fn decimal_data_type(meta: &NativeColumnMeta, default_precision: i32) -> ArrowDataType {
826 let (precision, scale) = decimal_precision_scale(meta, default_precision);
827 ArrowDataType::Decimal128(precision, scale)
828}
829
830fn fill_date_column(
831 data: &[u8],
832 offset: &mut usize,
833 name: &str,
834 num_rows: usize,
835) -> Result<BuiltColumn, TransportError> {
836 let mut builder = Date32Builder::with_capacity(num_rows);
837 for _ in 0..num_rows {
838 if read_u8(data, offset)? == 0 {
839 builder.append_null();
840 } else {
841 builder.append_value(read_packed_date_days(data, offset)?);
842 }
843 }
844 Ok((
845 Field::new(name, ArrowDataType::Date32, true),
846 Arc::new(builder.finish()),
847 ))
848}
849
850fn fill_timestamp_column(
851 data: &[u8],
852 offset: &mut usize,
853 meta: &NativeColumnMeta,
854 num_rows: usize,
855) -> Result<BuiltColumn, TransportError> {
856 let mut builder = TimestampMicrosecondBuilder::with_capacity(num_rows);
857 for _ in 0..num_rows {
858 if read_u8(data, offset)? == 0 {
859 builder.append_null();
860 } else {
861 builder.append_value(read_timestamp_micros(data, offset)?);
862 }
863 }
864
865 let finished = builder.finish();
866 if meta.type_id == T_TIMESTAMP_UTC {
867 return Ok((
868 Field::new(
869 &meta.name,
870 ArrowDataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
871 true,
872 ),
873 Arc::new(finished.with_timezone("UTC")),
874 ));
875 }
876 Ok((
877 Field::new(
878 &meta.name,
879 ArrowDataType::Timestamp(TimeUnit::Microsecond, None),
880 true,
881 ),
882 Arc::new(finished),
883 ))
884}
885
886fn read_packed_date_days(data: &[u8], offset: &mut usize) -> Result<i32, TransportError> {
888 let packed = read_i32(data, offset)?;
889 let year = packed >> 16;
890 let month = ((packed >> 8) & 0xFF) as u32;
891 let day = (packed & 0xFF) as u32;
892 Ok(crate::types::conversion::ymd_to_days(year, month, day))
893}
894
895fn read_timestamp_micros(data: &[u8], offset: &mut usize) -> Result<i64, TransportError> {
896 let year = read_i16(data, offset)? as i32;
897 let month = read_u8(data, offset)? as u32;
898 let day = read_u8(data, offset)? as u32;
899 let hour = read_u8(data, offset)? as u64;
900 let minute = read_u8(data, offset)? as u64;
901 let second = read_u8(data, offset)? as u64;
902 let nanos = read_i32(data, offset)?;
903 Ok(crate::types::conversion::ymd_hms_nanos_to_micros(
904 year, month, day, hour, minute, second, nanos,
905 ))
906}
907
908fn read_length_prefixed<'a>(
911 data: &'a [u8],
912 offset: &mut usize,
913 truncated_message: &str,
914) -> Result<&'a [u8], TransportError> {
915 let len = read_i32(data, offset)? as usize;
916 if *offset + len > data.len() {
917 return Err(TransportError::ProtocolError(truncated_message.to_owned()));
918 }
919 let value = &data[*offset..*offset + len];
920 *offset += len;
921 Ok(value)
922}
923
924fn fill_string_builder(
927 data: &[u8],
928 offset: &mut usize,
929 name: &str,
930 num_rows: usize,
931) -> Result<BuiltColumn, TransportError> {
932 let mut builder = StringBuilder::with_capacity(num_rows, num_rows * 16);
933 for _ in 0..num_rows {
934 if read_u8(data, offset)? == 0 {
935 builder.append_null();
936 } else {
937 let bytes = read_length_prefixed(data, offset, "String data truncated")?;
938 let text = std::str::from_utf8(bytes).map_err(|e| {
939 TransportError::ProtocolError(format!("Invalid UTF-8 in string: {}", e))
940 })?;
941 builder.append_value(text);
942 }
943 }
944 Ok((
945 Field::new(name, ArrowDataType::Utf8, true),
946 Arc::new(builder.finish()),
947 ))
948}
949
950pub fn parse_fetch_to_record_batch(
951 data: &[u8],
952 columns: &[NativeColumnMeta],
953) -> Result<(i64, arrow::record_batch::RecordBatch), TransportError> {
954 let mut offset = 0;
955 let rows_received = read_i64(data, &mut offset)?;
956 if rows_received < 0 {
957 return Err(TransportError::ProtocolError(format!(
958 "Invalid row count from server: {}",
959 rows_received
960 )));
961 }
962 let batch = build_batch_from_wire(data, &mut offset, columns, rows_received as usize)?;
963 Ok((rows_received, batch))
964}
965
966fn read_u8(data: &[u8], offset: &mut usize) -> Result<u8, TransportError> {
969 if *offset >= data.len() {
970 return Err(TransportError::ProtocolError("Data truncated (u8)".into()));
971 }
972 let v = data[*offset];
973 *offset += 1;
974 Ok(v)
975}
976
977fn read_i16(data: &[u8], offset: &mut usize) -> Result<i16, TransportError> {
978 if *offset + 2 > data.len() {
979 return Err(TransportError::ProtocolError("Data truncated (i16)".into()));
980 }
981 let v = i16::from_le_bytes([data[*offset], data[*offset + 1]]);
982 *offset += 2;
983 Ok(v)
984}
985
986fn read_i32(data: &[u8], offset: &mut usize) -> Result<i32, TransportError> {
987 if *offset + 4 > data.len() {
988 return Err(TransportError::ProtocolError("Data truncated (i32)".into()));
989 }
990 let v = i32::from_le_bytes([
991 data[*offset],
992 data[*offset + 1],
993 data[*offset + 2],
994 data[*offset + 3],
995 ]);
996 *offset += 4;
997 Ok(v)
998}
999
1000fn read_i64(data: &[u8], offset: &mut usize) -> Result<i64, TransportError> {
1001 if *offset + 8 > data.len() {
1002 return Err(TransportError::ProtocolError("Data truncated (i64)".into()));
1003 }
1004 let v = i64::from_le_bytes([
1005 data[*offset],
1006 data[*offset + 1],
1007 data[*offset + 2],
1008 data[*offset + 3],
1009 data[*offset + 4],
1010 data[*offset + 5],
1011 data[*offset + 6],
1012 data[*offset + 7],
1013 ]);
1014 *offset += 8;
1015 Ok(v)
1016}
1017
1018fn read_f64(data: &[u8], offset: &mut usize) -> Result<f64, TransportError> {
1019 if *offset + 8 > data.len() {
1020 return Err(TransportError::ProtocolError("Data truncated (f64)".into()));
1021 }
1022 let v = f64::from_le_bytes([
1023 data[*offset],
1024 data[*offset + 1],
1025 data[*offset + 2],
1026 data[*offset + 3],
1027 data[*offset + 4],
1028 data[*offset + 5],
1029 data[*offset + 6],
1030 data[*offset + 7],
1031 ]);
1032 *offset += 8;
1033 Ok(v)
1034}
1035
1036fn read_f32(data: &[u8], offset: &mut usize) -> Result<f32, TransportError> {
1037 if *offset + 4 > data.len() {
1038 return Err(TransportError::ProtocolError("Data truncated (f32)".into()));
1039 }
1040 let v = f32::from_le_bytes([
1041 data[*offset],
1042 data[*offset + 1],
1043 data[*offset + 2],
1044 data[*offset + 3],
1045 ]);
1046 *offset += 4;
1047 Ok(v)
1048}
1049
1050fn read_i128(data: &[u8], offset: &mut usize) -> Result<i128, TransportError> {
1051 if *offset + 16 > data.len() {
1052 return Err(TransportError::ProtocolError(
1053 "Data truncated (i128)".into(),
1054 ));
1055 }
1056 let v = i128::from_le_bytes([
1057 data[*offset],
1058 data[*offset + 1],
1059 data[*offset + 2],
1060 data[*offset + 3],
1061 data[*offset + 4],
1062 data[*offset + 5],
1063 data[*offset + 6],
1064 data[*offset + 7],
1065 data[*offset + 8],
1066 data[*offset + 9],
1067 data[*offset + 10],
1068 data[*offset + 11],
1069 data[*offset + 12],
1070 data[*offset + 13],
1071 data[*offset + 14],
1072 data[*offset + 15],
1073 ]);
1074 *offset += 16;
1075 Ok(v)
1076}
1077
1078#[cfg(test)]
1079mod tests {
1080 use super::super::arrow_builder::native_meta_to_data_type;
1081 use super::super::constants::{
1082 INTEGER_DECIMAL_PRECISION, IS_UTF8, SMALLINT_DECIMAL_PRECISION, SMALL_RESULTSET,
1083 };
1084 use super::*;
1085
1086 #[test]
1087 fn parse_empty_response() {
1088 let resp = parse_response(&[]).unwrap();
1089 assert!(matches!(resp.terminal, NativeResponse::Empty));
1090 }
1091
1092 #[test]
1093 fn parse_zero_result_count_response() {
1094 let resp = parse_response(&0i32.to_le_bytes()).unwrap();
1095 assert!(matches!(resp.terminal, NativeResponse::Empty));
1096 }
1097
1098 #[test]
1099 fn parse_result_count_with_still_executing() {
1100 let mut data = Vec::new();
1101 data.extend_from_slice(&1i32.to_le_bytes());
1102 data.push(R_STILL_EXECUTING as u8);
1103
1104 let resp = parse_response(&data).unwrap();
1105 assert!(matches!(resp.terminal, NativeResponse::StillExecuting));
1106 }
1107
1108 #[test]
1109 fn parse_result_count_with_row_count_response() {
1110 let mut data = Vec::new();
1111 data.extend_from_slice(&1i32.to_le_bytes());
1112 data.push(R_ROW_COUNT as u8);
1113 data.extend_from_slice(&42i64.to_le_bytes());
1114 let resp = parse_response(&data).unwrap();
1115 match resp.terminal {
1116 NativeResponse::RowCount(c) => assert_eq!(c, 42),
1117 _ => panic!("Expected RowCount"),
1118 }
1119 }
1120
1121 #[test]
1122 fn parse_result_count_with_exception_response() {
1123 let msg = "test error";
1124 let state = "42000";
1125 let mut data = Vec::new();
1126 data.extend_from_slice(&1i32.to_le_bytes());
1127 data.push(R_EXCEPTION as u8);
1128 data.extend_from_slice(&(msg.len() as i32).to_le_bytes());
1129 data.extend_from_slice(msg.as_bytes());
1130 data.extend_from_slice(state.as_bytes());
1131
1132 let resp = parse_response(&data).unwrap();
1133 match resp.terminal {
1134 NativeResponse::Exception { message, sql_state } => {
1135 assert_eq!(message, "test error");
1136 assert_eq!(sql_state, "42000");
1137 }
1138 _ => panic!("Expected Exception"),
1139 }
1140 }
1141
1142 fn write_col_meta(buf: &mut Vec<u8>, name: &str, type_id: u32, extra: &[u8]) {
1144 buf.extend_from_slice(&(name.len() as i32).to_le_bytes());
1145 buf.extend_from_slice(name.as_bytes());
1146 buf.extend_from_slice(&(type_id as i32).to_le_bytes());
1147 buf.extend_from_slice(extra);
1148 }
1149
1150 fn envelope(parts: &[Vec<u8>]) -> Vec<u8> {
1151 let mut data = Vec::new();
1152 data.extend_from_slice(&(parts.len() as i32).to_le_bytes());
1153 for part in parts {
1154 data.extend_from_slice(part);
1155 }
1156 data
1157 }
1158
1159 fn warning_part(message: &str, sql_state: &str) -> Vec<u8> {
1160 let mut data = Vec::new();
1161 data.push(R_WARNING as u8);
1162 data.extend_from_slice(&(message.len() as i32).to_le_bytes());
1163 data.extend_from_slice(message.as_bytes());
1164 data.extend_from_slice(sql_state.as_bytes());
1165 data
1166 }
1167
1168 fn row_count_part(count: i64) -> Vec<u8> {
1169 let mut data = Vec::new();
1170 data.push(R_ROW_COUNT as u8);
1171 data.extend_from_slice(&count.to_le_bytes());
1172 data
1173 }
1174
1175 fn empty_part() -> Vec<u8> {
1176 vec![R_EMPTY as u8]
1177 }
1178
1179 fn column_count_part(count: i32) -> Vec<u8> {
1180 let mut data = Vec::new();
1181 data.push(R_COLUMN_COUNT as u8);
1182 data.extend_from_slice(&count.to_le_bytes());
1183 data
1184 }
1185
1186 fn result_set_part(body: Vec<u8>) -> Vec<u8> {
1187 let mut data = Vec::with_capacity(body.len() + 1);
1188 data.push(R_RESULT_SET as u8);
1189 data.extend_from_slice(&body);
1190 data
1191 }
1192
1193 fn build_result_set_header(columns: &[(&str, u32, Vec<u8>)]) -> Vec<u8> {
1196 build_result_set_header_with_handle(SMALL_RESULTSET, columns)
1197 }
1198
1199 fn build_result_set_header_with_handle(
1200 handle: i32,
1201 columns: &[(&str, u32, Vec<u8>)],
1202 ) -> Vec<u8> {
1203 let mut data = Vec::new();
1204 data.extend_from_slice(&handle.to_le_bytes());
1205 data.extend_from_slice(&(columns.len() as i32).to_le_bytes());
1206 data.extend_from_slice(&0i64.to_le_bytes());
1208 data.extend_from_slice(&0i64.to_le_bytes());
1210 for (name, type_id, extra) in columns {
1211 write_col_meta(&mut data, name, *type_id, extra);
1212 }
1213 data
1214 }
1215
1216 fn exception_part(message: &str, sql_state: &str) -> Vec<u8> {
1217 let mut data = vec![R_EXCEPTION as u8];
1218 data.extend_from_slice(&(message.len() as i32).to_le_bytes());
1219 data.extend_from_slice(message.as_bytes());
1220 data.extend_from_slice(sql_state.as_bytes());
1221 data
1222 }
1223
1224 fn handle_part(statement_handle: i32, sub_results: &[Vec<u8>]) -> Vec<u8> {
1225 let mut data = vec![R_HANDLE as u8];
1226 data.extend_from_slice(&statement_handle.to_le_bytes());
1227 for sub in sub_results {
1228 data.extend_from_slice(sub);
1229 }
1230 data
1231 }
1232
1233 fn more_rows_part(payload: &[u8]) -> Vec<u8> {
1234 let mut data = vec![R_MORE_ROWS as u8];
1235 data.extend_from_slice(payload);
1236 data
1237 }
1238
1239 fn varchar_meta_bytes(max_len: i32) -> Vec<u8> {
1241 let mut extra = vec![IS_VARCHAR | IS_UTF8];
1242 extra.extend_from_slice(&max_len.to_le_bytes());
1243 extra.extend_from_slice(&(max_len * 4).to_le_bytes());
1244 extra
1245 }
1246
1247 fn protocol_message(error: TransportError) -> String {
1248 match error {
1249 TransportError::ProtocolError(msg) => msg,
1250 other => panic!("expected ProtocolError, got {other:?}"),
1251 }
1252 }
1253
1254 fn parse_error(data: &[u8]) -> String {
1255 protocol_message(parse_response(data).unwrap_err())
1256 }
1257
1258 fn legacy_error(data: &[u8]) -> String {
1259 protocol_message(parse_legacy_response(data).unwrap_err())
1260 }
1261
1262 #[test]
1263 fn parse_result_count_with_empty_response() {
1264 let resp = parse_response(&envelope(&[empty_part()])).unwrap();
1265 assert!(matches!(resp.terminal, NativeResponse::Empty));
1266 }
1267
1268 #[test]
1269 fn parse_warning_then_row_count_response() {
1270 let resp = parse_response(&envelope(&[
1271 warning_part("careful", "01000"),
1272 row_count_part(7),
1273 ]))
1274 .unwrap();
1275
1276 assert_eq!(resp.warnings.len(), 1);
1277 assert_eq!(resp.warnings[0].message, "careful");
1278 assert_eq!(resp.warnings[0].sql_state, "01000");
1279 assert!(matches!(resp.terminal, NativeResponse::RowCount(7)));
1280 }
1281
1282 #[test]
1283 fn parse_warning_then_result_set_response() {
1284 let resp = parse_response(&envelope(&[
1285 warning_part("heads up", "01000"),
1286 result_set_part(build_result_set_header(&[("ID", T_INTEGER, Vec::new())])),
1287 ]))
1288 .unwrap();
1289
1290 assert_eq!(resp.warnings.len(), 1);
1291 match resp.terminal {
1292 NativeResponse::ResultSet { columns, .. } => {
1293 assert_eq!(columns.len(), 1);
1294 assert_eq!(columns[0].name, "ID");
1295 }
1296 _ => panic!("Expected ResultSet"),
1297 }
1298 }
1299
1300 #[test]
1301 fn parse_column_count_then_empty_response() {
1302 let resp = parse_response(&envelope(&[column_count_part(3), empty_part()])).unwrap();
1303 assert!(matches!(resp.terminal, NativeResponse::Empty));
1304 }
1305
1306 #[test]
1307 fn parse_column_meta_timestamp_reads_precision() {
1308 let mut ts_extra = Vec::new();
1310 ts_extra.extend_from_slice(&3i32.to_le_bytes()); let columns = vec![("created_at", T_TIMESTAMP, ts_extra)];
1313 let data = build_result_set_header(&columns);
1314 let result = parse_result_set(&data).unwrap();
1315
1316 match result {
1317 NativeResponse::ResultSet { columns, .. } => {
1318 assert_eq!(columns.len(), 1);
1319 assert_eq!(columns[0].name, "created_at");
1320 assert_eq!(columns[0].type_id, T_TIMESTAMP);
1321 }
1322 _ => panic!("Expected ResultSet"),
1323 }
1324 }
1325
1326 #[test]
1327 fn parse_column_meta_timestamp_utc_reads_precision() {
1328 let mut ts_extra = Vec::new();
1329 ts_extra.extend_from_slice(&6i32.to_le_bytes()); let columns = vec![("ts_utc", T_TIMESTAMP_UTC, ts_extra)];
1332 let data = build_result_set_header(&columns);
1333 let result = parse_result_set(&data).unwrap();
1334
1335 match result {
1336 NativeResponse::ResultSet { columns, .. } => {
1337 assert_eq!(columns.len(), 1);
1338 assert_eq!(columns[0].name, "ts_utc");
1339 assert_eq!(columns[0].type_id, T_TIMESTAMP_UTC);
1340 }
1341 _ => panic!("Expected ResultSet"),
1342 }
1343 }
1344
1345 #[test]
1346 fn parse_column_meta_interval_year_reads_char_and_precision() {
1347 let mut extra = Vec::new();
1349 extra.push(0u8); extra.extend_from_slice(&13i32.to_le_bytes()); extra.extend_from_slice(&13i32.to_le_bytes()); extra.extend_from_slice(&2i32.to_le_bytes()); let columns = vec![("interval_col", T_INTERVAL_YEAR, extra)];
1355 let data = build_result_set_header(&columns);
1356 let result = parse_result_set(&data).unwrap();
1357
1358 match result {
1359 NativeResponse::ResultSet { columns, .. } => {
1360 assert_eq!(columns.len(), 1);
1361 assert_eq!(columns[0].name, "interval_col");
1362 assert_eq!(columns[0].type_id, T_INTERVAL_YEAR);
1363 assert_eq!(columns[0].interval_fraction, None);
1364 }
1365 _ => panic!("Expected ResultSet"),
1366 }
1367 }
1368
1369 #[test]
1370 fn parse_column_meta_interval_day_reads_char_and_two_precisions() {
1371 let mut extra = Vec::new();
1374 extra.push(0u8); extra.extend_from_slice(&29i32.to_le_bytes()); extra.extend_from_slice(&29i32.to_le_bytes()); extra.extend_from_slice(&2i32.to_le_bytes()); extra.extend_from_slice(&6i32.to_le_bytes()); let columns = vec![("interval_day_col", T_INTERVAL_DAY, extra)];
1381 let data = build_result_set_header(&columns);
1382 let result = parse_result_set(&data).unwrap();
1383
1384 match result {
1385 NativeResponse::ResultSet { columns, .. } => {
1386 assert_eq!(columns.len(), 1);
1387 assert_eq!(columns[0].name, "interval_day_col");
1388 assert_eq!(columns[0].type_id, T_INTERVAL_DAY);
1389 assert_eq!(columns[0].interval_fraction, Some(6));
1390 }
1391 _ => panic!("Expected ResultSet"),
1392 }
1393 }
1394
1395 #[test]
1396 fn parse_column_meta_geometry_reads_char_metadata() {
1397 let mut extra = Vec::new();
1399 extra.push(0u8); extra.extend_from_slice(&100i32.to_le_bytes()); extra.extend_from_slice(&100i32.to_le_bytes()); let columns = vec![("geom_col", T_GEOMETRY, extra)];
1404 let data = build_result_set_header(&columns);
1405 let result = parse_result_set(&data).unwrap();
1406
1407 match result {
1408 NativeResponse::ResultSet { columns, .. } => {
1409 assert_eq!(columns.len(), 1);
1410 assert_eq!(columns[0].name, "geom_col");
1411 assert_eq!(columns[0].type_id, T_GEOMETRY);
1412 }
1413 _ => panic!("Expected ResultSet"),
1414 }
1415 }
1416
1417 #[test]
1418 fn varchar_flag_bit_discriminates_varchar_from_char() {
1419 let reported_type = |vc_flag: u8| {
1422 let mut extra = vec![vc_flag];
1423 extra.extend_from_slice(&50i32.to_le_bytes()); extra.extend_from_slice(&200i32.to_le_bytes()); let mut data = Vec::new();
1426 write_col_meta(&mut data, "NAME", T_CHAR, &extra);
1427 let mut offset = 0;
1428 native_meta_to_data_type(&parse_column_meta(&data, &mut offset).unwrap())
1429 };
1430
1431 let varchar = reported_type(0x11);
1432 assert_eq!(varchar.type_name, "VARCHAR");
1433 assert_eq!(varchar.size, Some(50));
1434
1435 let char_utf8 = reported_type(0x10);
1436 assert_eq!(char_utf8.type_name, "CHAR");
1437 assert_eq!(char_utf8.size, Some(50));
1438
1439 assert_ne!(reported_type(0x00).type_name, "VARCHAR");
1440 }
1441
1442 #[test]
1443 fn parse_multi_column_benchmark_table_metadata() {
1444 let mut columns = Vec::new();
1450
1451 let mut decimal_extra = Vec::new();
1453 decimal_extra.extend_from_slice(&18i32.to_le_bytes()); decimal_extra.extend_from_slice(&0i32.to_le_bytes()); columns.push(("ID", T_DECIMAL, decimal_extra));
1456
1457 let mut vc_extra = Vec::new();
1459 vc_extra.push(IS_VARCHAR | IS_UTF8); vc_extra.extend_from_slice(&100i32.to_le_bytes()); vc_extra.extend_from_slice(&400i32.to_le_bytes()); columns.push(("NAME", T_CHAR, vc_extra));
1463
1464 let mut vc_extra2 = Vec::new();
1466 vc_extra2.push(IS_VARCHAR | IS_UTF8);
1467 vc_extra2.extend_from_slice(&200i32.to_le_bytes());
1468 vc_extra2.extend_from_slice(&800i32.to_le_bytes());
1469 columns.push(("EMAIL", T_CHAR, vc_extra2));
1470
1471 let mut int_extra = Vec::new();
1473 int_extra.extend_from_slice(&18i32.to_le_bytes());
1474 int_extra.extend_from_slice(&0i32.to_le_bytes());
1475 columns.push(("AGE", T_DECIMAL, int_extra));
1476
1477 let mut dec_extra = Vec::new();
1479 dec_extra.extend_from_slice(&12i32.to_le_bytes());
1480 dec_extra.extend_from_slice(&2i32.to_le_bytes());
1481 columns.push(("SALARY", T_DECIMAL, dec_extra));
1482
1483 let mut ts_extra = Vec::new();
1485 ts_extra.extend_from_slice(&3i32.to_le_bytes()); columns.push(("CREATED_AT", T_TIMESTAMP, ts_extra));
1487
1488 columns.push(("IS_ACTIVE", T_BOOLEAN, Vec::new()));
1490
1491 let mut vc_extra3 = Vec::new();
1493 vc_extra3.push(IS_VARCHAR | IS_UTF8);
1494 vc_extra3.extend_from_slice(&1000i32.to_le_bytes());
1495 vc_extra3.extend_from_slice(&4000i32.to_le_bytes());
1496 columns.push(("DESCRIPTION", T_CHAR, vc_extra3));
1497
1498 let data = build_result_set_header(&columns);
1499 let result = parse_result_set(&data).unwrap();
1500
1501 match result {
1502 NativeResponse::ResultSet {
1503 columns: cols,
1504 total_rows,
1505 rows_received,
1506 ..
1507 } => {
1508 assert_eq!(cols.len(), 8);
1509 assert_eq!(cols[0].name, "ID");
1510 assert_eq!(cols[0].type_id, T_DECIMAL);
1511 assert_eq!(cols[1].name, "NAME");
1512 assert_eq!(cols[1].type_id, T_CHAR);
1513 assert!(cols[1].is_varchar);
1514 assert_eq!(cols[2].name, "EMAIL");
1515 assert_eq!(cols[3].name, "AGE");
1516 assert_eq!(cols[4].name, "SALARY");
1517 assert_eq!(cols[4].precision, Some(12));
1518 assert_eq!(cols[4].scale, Some(2));
1519 assert_eq!(cols[5].name, "CREATED_AT");
1520 assert_eq!(cols[5].type_id, T_TIMESTAMP);
1521 assert_eq!(cols[6].name, "IS_ACTIVE");
1522 assert_eq!(cols[6].type_id, T_BOOLEAN);
1523 assert_eq!(cols[7].name, "DESCRIPTION");
1524 assert_eq!(cols[7].type_id, T_CHAR);
1525 assert!(cols[7].is_varchar);
1526 assert_eq!(total_rows, 0);
1527 assert_eq!(rows_received, 0);
1528 }
1529 _ => panic!("Expected ResultSet"),
1530 }
1531 }
1532
1533 fn meta(
1534 name: &str,
1535 type_id: u32,
1536 precision: Option<i32>,
1537 scale: Option<i32>,
1538 ) -> NativeColumnMeta {
1539 NativeColumnMeta {
1540 name: name.to_string(),
1541 type_id,
1542 precision,
1543 scale,
1544 is_varchar: false,
1545 max_len: None,
1546 interval_fraction: None,
1547 }
1548 }
1549
1550 #[test]
1551 fn single_pass_date_column_decodes_to_date32() {
1552 use arrow::array::{Array, Date32Array};
1553 let columns = vec![meta("d", T_DATE, None, None)];
1554 let mut data = Vec::new();
1555 let packed = (2024i32 << 16) | (1 << 8) | 2;
1557 data.push(1u8);
1558 data.extend_from_slice(&packed.to_le_bytes());
1559 data.push(0u8);
1561
1562 let mut offset = 0;
1563 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
1564 assert_eq!(batch.num_rows(), 2);
1565 assert_eq!(batch.num_columns(), 1);
1566
1567 let arr = batch
1568 .column(0)
1569 .as_any()
1570 .downcast_ref::<Date32Array>()
1571 .unwrap();
1572 assert!(arr.is_null(1));
1573 assert_eq!(arr.value(0), 19724);
1574 assert_eq!(offset, data.len());
1575 }
1576
1577 #[test]
1579 fn single_pass_dates_and_timestamps_before_1970_keep_their_calendar_day() {
1580 use arrow::array::{Date32Array, TimestampMicrosecondArray};
1581
1582 let date_columns = vec![meta("d", T_DATE, None, None)];
1583 let mut date_data = Vec::new();
1584 for (year, month, day) in [(1968i32, 1i32, 1i32), (1, 1, 1)] {
1585 date_data.push(1u8);
1586 date_data.extend_from_slice(&((year << 16) | (month << 8) | day).to_le_bytes());
1587 }
1588 let mut offset = 0;
1589 let batch = build_batch_from_wire(&date_data, &mut offset, &date_columns, 2).unwrap();
1590 let dates = batch
1591 .column(0)
1592 .as_any()
1593 .downcast_ref::<Date32Array>()
1594 .unwrap();
1595 assert_eq!(dates.value(0), -731);
1596 assert_eq!(dates.value(1), -719162);
1597
1598 let ts_columns = vec![meta("ts", T_TIMESTAMP, None, None)];
1599 let mut ts_data = Vec::new();
1600 for (year, month, day, hour, minute, second, nanos) in [
1601 (1950i16, 6u8, 15u8, 0u8, 0u8, 0u8, 0i32),
1602 (1969, 12, 31, 23, 59, 59, 999_999_000),
1603 ] {
1604 ts_data.push(1u8);
1605 ts_data.extend_from_slice(&year.to_le_bytes());
1606 ts_data.extend_from_slice(&[month, day, hour, minute, second]);
1607 ts_data.extend_from_slice(&nanos.to_le_bytes());
1608 }
1609 let mut offset = 0;
1610 let batch = build_batch_from_wire(&ts_data, &mut offset, &ts_columns, 2).unwrap();
1611 let timestamps = batch
1612 .column(0)
1613 .as_any()
1614 .downcast_ref::<TimestampMicrosecondArray>()
1615 .unwrap();
1616 assert_eq!(timestamps.value(0), -616_896_000_000_000);
1617 assert_eq!(timestamps.value(1), -1);
1618 }
1619
1620 #[test]
1621 fn single_pass_timestamp_column_decodes_to_microseconds() {
1622 use arrow::array::{Array, TimestampMicrosecondArray};
1623
1624 let columns = vec![meta("ts", T_TIMESTAMP, None, None)];
1625 let mut data = Vec::new();
1626 data.push(1u8);
1628 data.extend_from_slice(&1970i16.to_le_bytes());
1629 data.push(1u8);
1630 data.push(1u8);
1631 data.push(0u8);
1632 data.push(0u8);
1633 data.push(0u8);
1634 data.extend_from_slice(&123_000_000i32.to_le_bytes()); data.push(0u8);
1637
1638 let mut offset = 0;
1639 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
1640 let arr = batch
1641 .column(0)
1642 .as_any()
1643 .downcast_ref::<TimestampMicrosecondArray>()
1644 .unwrap();
1645 assert_eq!(arr.value(0), 123_000);
1646 assert!(arr.is_null(1));
1647 }
1648
1649 #[test]
1650 fn single_pass_timestamp_utc_has_utc_timezone() {
1651 use arrow::array::Array;
1652 use arrow::datatypes::{DataType, TimeUnit};
1653
1654 let columns = vec![meta("ts", T_TIMESTAMP_UTC, None, None)];
1655 let mut data = Vec::new();
1656 data.push(1u8);
1657 data.extend_from_slice(&1970i16.to_le_bytes());
1658 data.push(1u8);
1659 data.push(1u8);
1660 data.push(0u8);
1661 data.push(0u8);
1662 data.push(0u8);
1663 data.extend_from_slice(&0i32.to_le_bytes());
1664
1665 let mut offset = 0;
1666 let batch = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap();
1667 let arr = batch.column(0);
1668 assert_eq!(
1669 arr.data_type(),
1670 &DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()))
1671 );
1672 }
1673
1674 #[test]
1675 fn single_pass_string_column_decodes_utf8_without_owning_string() {
1676 use arrow::array::{Array, StringArray};
1677
1678 let columns = vec![meta("s", T_CHAR, None, None)];
1679 let mut data = Vec::new();
1680 data.push(1u8);
1682 data.extend_from_slice(&5i32.to_le_bytes());
1683 data.extend_from_slice(b"hello");
1684 data.push(0u8);
1686
1687 let mut offset = 0;
1688 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
1689 let arr = batch
1690 .column(0)
1691 .as_any()
1692 .downcast_ref::<StringArray>()
1693 .unwrap();
1694 assert_eq!(arr.value(0), "hello");
1695 assert!(arr.is_null(1));
1696 }
1697
1698 #[test]
1699 fn single_pass_all_primitive_types_offsets_align() {
1700 let columns = vec![
1702 meta("d", T_DOUBLE, None, None),
1703 meta("b", T_BOOLEAN, None, None),
1704 meta("i", T_INTEGER, None, None),
1705 ];
1706 let mut data = Vec::new();
1707 data.push(1u8);
1709 data.extend_from_slice(&1.5f64.to_le_bytes());
1710 data.push(0u8); data.push(1u8);
1714 data.push(1u8); data.push(0u8); data.push(1u8);
1719 data.extend_from_slice(&42i64.to_le_bytes());
1720 data.push(0u8); let mut offset = 0;
1723 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
1724 assert_eq!(offset, data.len(), "all bytes should be consumed");
1725 assert_eq!(batch.num_rows(), 2);
1726 assert_eq!(batch.num_columns(), 3);
1727
1728 use arrow::array::{Array, BooleanArray, Decimal128Array, Float64Array};
1729 let f = batch
1730 .column(0)
1731 .as_any()
1732 .downcast_ref::<Float64Array>()
1733 .unwrap();
1734 let b = batch
1735 .column(1)
1736 .as_any()
1737 .downcast_ref::<BooleanArray>()
1738 .unwrap();
1739 let i = batch
1740 .column(2)
1741 .as_any()
1742 .downcast_ref::<Decimal128Array>()
1743 .unwrap();
1744
1745 assert_eq!(f.value(0), 1.5);
1746 assert!(f.is_null(1));
1747 assert!(b.value(0));
1748 assert!(b.is_null(1));
1749 assert_eq!(i.value(0), 42);
1750 assert!(i.is_null(1));
1751 }
1752
1753 #[test]
1754 fn single_pass_empty_rows_produces_typed_empty_batch() {
1755 let columns = vec![
1756 meta("i", T_INTEGER, None, None),
1757 meta("s", T_CHAR, None, None),
1758 ];
1759 let data = Vec::new();
1760 let mut offset = 0;
1761 let batch = build_batch_from_wire(&data, &mut offset, &columns, 0).unwrap();
1762 assert_eq!(batch.num_rows(), 0);
1763 assert_eq!(batch.num_columns(), 2);
1764 use arrow::datatypes::DataType;
1765 assert_eq!(
1766 batch.schema().field(0).data_type(),
1767 &DataType::Decimal128(18, 0)
1768 );
1769 assert_eq!(batch.schema().field(1).data_type(), &DataType::Utf8);
1770 }
1771
1772 #[test]
1773 fn single_pass_binary_column_copies_into_builder() {
1774 use arrow::array::{Array, BinaryArray};
1775 let columns = vec![meta("bin", T_BINARY, None, None)];
1776 let mut data = Vec::new();
1777 data.push(1u8);
1778 data.extend_from_slice(&3i32.to_le_bytes());
1779 data.extend_from_slice(&[0xDE, 0xAD, 0xBE]);
1780 data.push(0u8);
1781
1782 let mut offset = 0;
1783 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
1784 let arr = batch
1785 .column(0)
1786 .as_any()
1787 .downcast_ref::<BinaryArray>()
1788 .unwrap();
1789 assert_eq!(arr.value(0), &[0xDE, 0xAD, 0xBE][..]);
1790 assert!(arr.is_null(1));
1791 }
1792
1793 #[test]
1794 fn parse_fetch_to_record_batch_reads_row_count_prefix() {
1795 use arrow::array::{Array, Decimal128Array};
1796 let columns = vec![meta("i", T_INTEGER, None, None)];
1797 let mut data = Vec::new();
1798 data.extend_from_slice(&2i64.to_le_bytes());
1800 data.push(1u8);
1802 data.extend_from_slice(&100i64.to_le_bytes());
1803 data.push(0u8);
1805
1806 let (rows, batch) = parse_fetch_to_record_batch(&data, &columns).unwrap();
1807 assert_eq!(rows, 2);
1808 let arr = batch
1809 .column(0)
1810 .as_any()
1811 .downcast_ref::<Decimal128Array>()
1812 .unwrap();
1813 assert_eq!(arr.value(0), 100);
1814 assert!(arr.is_null(1));
1815 }
1816
1817 #[test]
1818 fn single_pass_decimal_scale_zero_maps_to_decimal128() {
1819 use arrow::array::Decimal128Array;
1820 use arrow::datatypes::DataType;
1821
1822 let columns = vec![meta("big", T_DECIMAL, Some(18), Some(0))];
1823 let mut data = Vec::new();
1824 data.push(1u8);
1825 data.extend_from_slice(&12345i64.to_le_bytes());
1826
1827 let mut offset = 0;
1828 let batch = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap();
1829 assert_eq!(
1830 batch.schema().field(0).data_type(),
1831 &DataType::Decimal128(18, 0)
1832 );
1833 let arr = batch
1834 .column(0)
1835 .as_any()
1836 .downcast_ref::<Decimal128Array>()
1837 .unwrap();
1838 assert_eq!(arr.value(0), 12345i128);
1839 }
1840
1841 #[test]
1842 fn single_pass_decimal_with_scale_maps_to_decimal128_preserving_precision() {
1843 use arrow::array::Decimal128Array;
1844 use arrow::datatypes::DataType;
1845
1846 let columns = vec![meta("s", T_DECIMAL, Some(12), Some(2))];
1847 let mut data = Vec::new();
1848 data.push(1u8);
1850 data.extend_from_slice(&12345i64.to_le_bytes());
1851
1852 let mut offset = 0;
1853 let batch = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap();
1854 assert_eq!(
1855 batch.schema().field(0).data_type(),
1856 &DataType::Decimal128(12, 2)
1857 );
1858 let arr = batch
1859 .column(0)
1860 .as_any()
1861 .downcast_ref::<Decimal128Array>()
1862 .unwrap();
1863 assert_eq!(arr.value(0), 12345i128);
1864 assert_eq!(arr.value_as_string(0), "123.45");
1865 }
1866
1867 #[test]
1869 fn single_pass_smalldecimal_uses_i32_wire_and_decimal128() {
1870 use arrow::array::{Array, Decimal128Array};
1871 use arrow::datatypes::DataType;
1872
1873 let columns = vec![meta("s", T_SMALLDECIMAL, Some(5), Some(2))];
1874 let mut data = Vec::new();
1875 data.push(1u8);
1876 data.extend_from_slice(&12345i32.to_le_bytes());
1877 data.push(0u8); let mut offset = 0;
1880 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
1881 assert_eq!(
1882 batch.schema().field(0).data_type(),
1883 &DataType::Decimal128(5, 2)
1884 );
1885 let arr = batch
1886 .column(0)
1887 .as_any()
1888 .downcast_ref::<Decimal128Array>()
1889 .unwrap();
1890 assert_eq!(arr.value(0), 12345i128);
1891 assert!(arr.is_null(1));
1892 assert_eq!(offset, data.len());
1893 }
1894
1895 #[test]
1896 fn single_pass_decimal_precision_le_9_uses_i32_wire() {
1897 use arrow::array::Decimal128Array;
1898 use arrow::datatypes::DataType;
1899
1900 let columns = vec![meta("d", T_DECIMAL, Some(9), Some(3))];
1901 let mut data = Vec::new();
1902 data.push(1u8);
1903 data.extend_from_slice(&1_234_567i32.to_le_bytes());
1904
1905 let mut offset = 0;
1906 let batch = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap();
1907 assert_eq!(
1908 batch.schema().field(0).data_type(),
1909 &DataType::Decimal128(9, 3)
1910 );
1911 let arr = batch
1912 .column(0)
1913 .as_any()
1914 .downcast_ref::<Decimal128Array>()
1915 .unwrap();
1916 assert_eq!(arr.value(0), 1_234_567i128);
1917 assert_eq!(offset, data.len());
1918 }
1919
1920 #[test]
1921 fn single_pass_real_column_widens_f32_to_f64() {
1922 use arrow::array::{Array, Float64Array};
1923 use arrow::datatypes::DataType;
1924
1925 let columns = vec![meta("r", T_REAL, None, None)];
1926 let mut data = Vec::new();
1927 data.push(1u8);
1928 data.extend_from_slice(&1.5f32.to_le_bytes());
1929 data.push(0u8);
1930
1931 let mut offset = 0;
1932 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
1933 assert_eq!(batch.schema().field(0).data_type(), &DataType::Float64);
1934 let arr = batch
1935 .column(0)
1936 .as_any()
1937 .downcast_ref::<Float64Array>()
1938 .unwrap();
1939 assert_eq!(arr.value(0), 1.5);
1940 assert!(arr.is_null(1));
1941 assert_eq!(offset, data.len());
1942 }
1943
1944 #[test]
1946 fn single_pass_smallint_column_decodes_to_decimal128() {
1947 use arrow::array::{Array, Decimal128Array};
1948 use arrow::datatypes::DataType;
1949
1950 let columns = vec![meta("si", T_SMALLINT, None, None)];
1951 let mut data = Vec::new();
1952 for value in [-7i32, 999_999_999, -999_999_999] {
1953 data.push(1u8);
1954 data.extend_from_slice(&value.to_le_bytes());
1955 }
1956 data.push(0u8);
1957
1958 let mut offset = 0;
1959 let batch = build_batch_from_wire(&data, &mut offset, &columns, 4).unwrap();
1960 assert_eq!(
1961 batch.schema().field(0).data_type(),
1962 &DataType::Decimal128(9, 0)
1963 );
1964 let arr = batch
1965 .column(0)
1966 .as_any()
1967 .downcast_ref::<Decimal128Array>()
1968 .unwrap();
1969 assert_eq!(arr.value(0), -7);
1970 assert_eq!(arr.value(1), 999_999_999);
1971 assert_eq!(arr.value(2), -999_999_999);
1972 assert!(arr.is_null(3));
1973 assert_eq!(offset, data.len());
1974 }
1975
1976 #[test]
1978 fn single_pass_integer_column_decodes_to_decimal128() {
1979 use arrow::array::{Array, Decimal128Array};
1980 use arrow::datatypes::DataType;
1981
1982 let columns = vec![meta("i", T_INTEGER, None, None)];
1983 let mut data = Vec::new();
1984 for value in [999_999_999_999_999_999i64, -999_999_999_999_999_999] {
1985 data.push(1u8);
1986 data.extend_from_slice(&value.to_le_bytes());
1987 }
1988 data.push(0u8);
1989
1990 let mut offset = 0;
1991 let batch = build_batch_from_wire(&data, &mut offset, &columns, 3).unwrap();
1992 assert_eq!(
1993 batch.schema().field(0).data_type(),
1994 &DataType::Decimal128(18, 0)
1995 );
1996 let arr = batch
1997 .column(0)
1998 .as_any()
1999 .downcast_ref::<Decimal128Array>()
2000 .unwrap();
2001 assert_eq!(arr.value(0), 999_999_999_999_999_999);
2002 assert_eq!(arr.value(1), -999_999_999_999_999_999);
2003 assert!(arr.is_null(2));
2004 assert_eq!(offset, data.len());
2005 }
2006
2007 #[test]
2009 fn integer_wire_types_take_the_arrow_type_of_their_reported_decimal() {
2010 use arrow::datatypes::DataType;
2011
2012 for (type_id, precision) in [
2013 (T_INTEGER, INTEGER_DECIMAL_PRECISION),
2014 (T_SMALLINT, SMALLINT_DECIMAL_PRECISION),
2015 ] {
2016 let column = meta("n", type_id, None, None);
2017 let reported = native_meta_to_data_type(&column);
2018 assert_eq!(reported.type_name, "DECIMAL");
2019 assert_eq!(reported.precision, Some(precision));
2020 assert_eq!(reported.scale, Some(0));
2021 let batch = build_batch_from_wire(&[], &mut 0, &[column], 0).unwrap();
2022 assert_eq!(
2023 batch.schema().field(0).data_type(),
2024 &DataType::Decimal128(precision as u8, 0)
2025 );
2026 }
2027 }
2028
2029 #[test]
2031 fn integer_wire_types_ignore_a_precision_in_the_column_description() {
2032 use arrow::array::{Array, Decimal128Array};
2033 use arrow::datatypes::DataType;
2034
2035 let mut integer_data = vec![1u8];
2036 integer_data.extend_from_slice(&999_999_999_999_999_999i64.to_le_bytes());
2037 integer_data.push(0u8);
2038 let mut smallint_data = vec![1u8];
2039 smallint_data.extend_from_slice(&999_999_999i32.to_le_bytes());
2040 smallint_data.push(0u8);
2041 let cases = [
2042 (
2043 meta("i", T_INTEGER, Some(9), Some(2)),
2044 integer_data,
2045 DataType::Decimal128(18, 0),
2046 999_999_999_999_999_999i128,
2047 ),
2048 (
2049 meta("s", T_SMALLINT, Some(4), Some(2)),
2050 smallint_data,
2051 DataType::Decimal128(9, 0),
2052 999_999_999i128,
2053 ),
2054 ];
2055
2056 for (column, data, expected_type, expected_value) in cases {
2057 let mut offset = 0;
2058 let batch = build_batch_from_wire(&data, &mut offset, &[column], 2).unwrap();
2059 assert_eq!(batch.schema().field(0).data_type(), &expected_type);
2060 let arr = batch
2061 .column(0)
2062 .as_any()
2063 .downcast_ref::<Decimal128Array>()
2064 .unwrap();
2065 assert_eq!(arr.value(0), expected_value);
2066 assert!(arr.is_null(1));
2067 assert_eq!(offset, data.len());
2068 }
2069 }
2070
2071 #[test]
2072 fn single_pass_timestamp_with_local_time_zone_has_no_arrow_timezone() {
2073 use arrow::datatypes::{DataType, TimeUnit};
2074
2075 let columns = vec![meta("ts", T_TIMESTAMP_LOCAL_TZ, None, None)];
2076 let mut data = Vec::new();
2077 data.push(1u8);
2078 data.extend_from_slice(&1970i16.to_le_bytes());
2079 data.push(1u8);
2080 data.push(1u8);
2081 data.push(0u8);
2082 data.push(0u8);
2083 data.push(0u8);
2084 data.extend_from_slice(&0i32.to_le_bytes());
2085
2086 let mut offset = 0;
2087 let batch = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap();
2088 assert_eq!(
2089 batch.schema().field(0).data_type(),
2090 &DataType::Timestamp(TimeUnit::Microsecond, None)
2091 );
2092 }
2093
2094 #[test]
2095 fn single_pass_text_types_without_an_arrow_mapping_decode_as_utf8() {
2096 use arrow::array::StringArray;
2097 use arrow::datatypes::DataType;
2098
2099 const UNKNOWN_TYPE_ID: u32 = 9_999;
2100 let string_like = [T_CHAR, UNKNOWN_TYPE_ID];
2101
2102 for type_id in string_like {
2103 let columns = vec![meta("s", type_id, None, None)];
2104 let mut data = Vec::new();
2105 data.push(1u8);
2106 data.extend_from_slice(&2i32.to_le_bytes());
2107 data.extend_from_slice(b"ok");
2108
2109 let mut offset = 0;
2110 let batch = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap();
2111 assert_eq!(
2112 batch.schema().field(0).data_type(),
2113 &DataType::Utf8,
2114 "type_id {type_id}"
2115 );
2116 let arr = batch
2117 .column(0)
2118 .as_any()
2119 .downcast_ref::<StringArray>()
2120 .unwrap();
2121 assert_eq!(arr.value(0), "ok", "type_id {type_id}");
2122 assert_eq!(offset, data.len(), "type_id {type_id}");
2123 }
2124 }
2125
2126 fn length_prefixed_values(values: &[Option<&str>]) -> Vec<u8> {
2127 let mut data = Vec::new();
2128 for value in values {
2129 match value {
2130 Some(text) => {
2131 data.push(1u8);
2132 data.extend_from_slice(&(text.len() as i32).to_le_bytes());
2133 data.extend_from_slice(text.as_bytes());
2134 }
2135 None => data.push(0u8),
2136 }
2137 }
2138 data
2139 }
2140
2141 #[test]
2143 fn single_pass_interval_columns_decode_to_month_day_nano() {
2144 use arrow::array::{Array, IntervalMonthDayNanoArray};
2145 use arrow::datatypes::{DataType, IntervalMonthDayNano, IntervalUnit};
2146
2147 let expected = [
2148 (
2149 T_INTERVAL_YEAR,
2150 "+01-02",
2151 IntervalMonthDayNano::new(14, 0, 0),
2152 ),
2153 (
2154 T_INTERVAL_DAY,
2155 "-03 04:05:06.789",
2156 IntervalMonthDayNano::new(0, -3, -14_706_789_000_000),
2157 ),
2158 ];
2159 for (type_id, text, interval) in expected {
2160 let columns = vec![meta("i", type_id, None, None)];
2161 let data = length_prefixed_values(&[Some(text), None]);
2162
2163 let mut offset = 0;
2164 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
2165
2166 assert_eq!(
2167 batch.schema().field(0).data_type(),
2168 &DataType::Interval(IntervalUnit::MonthDayNano)
2169 );
2170 let values = batch
2171 .column(0)
2172 .as_any()
2173 .downcast_ref::<IntervalMonthDayNanoArray>()
2174 .unwrap();
2175 assert_eq!(values.value(0), interval, "{text}");
2176 assert!(values.is_null(1));
2177 assert_eq!(offset, data.len());
2178 }
2179 }
2180
2181 #[test]
2183 fn single_pass_interval_month_count_outside_i32_names_the_column() {
2184 let columns = vec![meta("BIG", T_INTERVAL_YEAR, None, None)];
2185 let data = length_prefixed_values(&[Some("+999999999-11")]);
2186
2187 let mut offset = 0;
2188 let error = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2189
2190 let message = error.to_string();
2191 assert!(message.contains("BIG"), "{message}");
2192 assert!(message.contains("+999999999-11"), "{message}");
2193 }
2194
2195 #[test]
2197 fn single_pass_interval_text_that_is_not_an_interval_names_the_column() {
2198 let columns = vec![meta("IV", T_INTERVAL_DAY, None, None)];
2199 let data = length_prefixed_values(&[Some("garbage")]);
2200
2201 let mut offset = 0;
2202 let error = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2203
2204 assert!(error.to_string().contains("IV"), "{error}");
2205 }
2206
2207 #[test]
2209 fn single_pass_geometry_and_hashtype_columns_keep_their_text_bytes_in_binary() {
2210 use arrow::array::{Array, BinaryArray};
2211 use arrow::datatypes::DataType;
2212
2213 let expected = [
2214 (T_GEOMETRY, "POINT (1 2)"),
2215 (T_HASHTYPE, "550e8400e29b11d4a716446655440000"),
2216 ];
2217 for (type_id, text) in expected {
2218 let columns = vec![meta("b", type_id, None, None)];
2219 let data = length_prefixed_values(&[Some(text), None]);
2220
2221 let mut offset = 0;
2222 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
2223
2224 assert_eq!(batch.schema().field(0).data_type(), &DataType::Binary);
2225 let values = batch
2226 .column(0)
2227 .as_any()
2228 .downcast_ref::<BinaryArray>()
2229 .unwrap();
2230 assert_eq!(values.value(0), text.as_bytes());
2231 assert!(values.is_null(1));
2232 assert_eq!(offset, data.len());
2233 }
2234 }
2235
2236 #[test]
2238 fn single_pass_bigdecimal_preserves_i128() {
2239 use arrow::array::Decimal128Array;
2240 use arrow::datatypes::DataType;
2241
2242 let big: i128 = (i64::MAX as i128) + 1;
2244 let columns = vec![meta("bd", T_BIGDECIMAL, Some(36), Some(0))];
2245 let mut data = Vec::new();
2246 data.push(1u8);
2247 data.extend_from_slice(&big.to_le_bytes());
2248
2249 let mut offset = 0;
2250 let batch = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap();
2251 assert_eq!(
2252 batch.schema().field(0).data_type(),
2253 &DataType::Decimal128(36, 0)
2254 );
2255 let arr = batch
2256 .column(0)
2257 .as_any()
2258 .downcast_ref::<Decimal128Array>()
2259 .unwrap();
2260 assert_eq!(arr.value(0), big);
2261 assert_eq!(offset, data.len());
2262 }
2263
2264 #[test]
2267 fn legacy_result_set_carries_its_column_metadata() {
2268 let data = result_set_part(build_result_set_header(&[("ID", T_INTEGER, Vec::new())]));
2269
2270 let resp = parse_legacy_response(&data).unwrap();
2271
2272 assert!(resp.warnings.is_empty());
2273 match resp.terminal {
2274 NativeResponse::ResultSet {
2275 handle, columns, ..
2276 } => {
2277 assert_eq!(handle, SMALL_RESULTSET);
2278 assert_eq!(columns.len(), 1);
2279 assert_eq!(columns[0].name, "ID");
2280 }
2281 other => panic!("Expected ResultSet, got {other:?}"),
2282 }
2283 }
2284
2285 #[test]
2286 fn legacy_handle_without_sub_result_reports_no_columns() {
2287 let resp = parse_legacy_response(&handle_part(77, &[])).unwrap();
2288
2289 match resp.terminal {
2290 NativeResponse::PreparedStatement {
2291 handle,
2292 parameters,
2293 result_columns,
2294 } => {
2295 assert_eq!(handle, 77);
2296 assert!(parameters.is_empty());
2297 assert!(result_columns.is_empty());
2298 }
2299 other => panic!("Expected PreparedStatement, got {other:?}"),
2300 }
2301 }
2302
2303 #[test]
2304 fn legacy_row_count_returns_the_affected_row_count() {
2305 let resp = parse_legacy_response(&row_count_part(9)).unwrap();
2306 assert!(matches!(resp.terminal, NativeResponse::RowCount(9)));
2307 }
2308
2309 #[test]
2310 fn legacy_exception_returns_message_and_sql_state() {
2311 let resp = parse_legacy_response(&exception_part("boom", "42000")).unwrap();
2312
2313 match resp.terminal {
2314 NativeResponse::Exception { message, sql_state } => {
2315 assert_eq!(message, "boom");
2316 assert_eq!(sql_state, "42000");
2317 }
2318 other => panic!("Expected Exception, got {other:?}"),
2319 }
2320 }
2321
2322 #[test]
2323 fn legacy_warning_is_collected_and_leaves_an_empty_terminal() {
2324 let resp = parse_legacy_response(&warning_part("careful", "01000")).unwrap();
2325
2326 assert_eq!(
2327 resp.warnings,
2328 vec![NativeWarning {
2329 message: "careful".to_owned(),
2330 sql_state: "01000".to_owned(),
2331 }]
2332 );
2333 assert!(matches!(resp.terminal, NativeResponse::Empty));
2334 }
2335
2336 #[test]
2337 fn legacy_column_count_is_consumed_and_yields_an_empty_terminal() {
2338 let resp = parse_legacy_response(&column_count_part(4)).unwrap();
2339
2340 assert!(resp.warnings.is_empty());
2341 assert!(matches!(resp.terminal, NativeResponse::Empty));
2342 }
2343
2344 #[test]
2345 fn legacy_empty_response_yields_an_empty_terminal() {
2346 let resp = parse_legacy_response(&empty_part()).unwrap();
2347 assert!(matches!(resp.terminal, NativeResponse::Empty));
2348 }
2349
2350 #[test]
2351 fn legacy_still_executing_without_status_message_is_accepted() {
2352 let resp = parse_legacy_response(&[R_STILL_EXECUTING as u8]).unwrap();
2353 assert!(matches!(resp.terminal, NativeResponse::StillExecuting));
2354 }
2355
2356 #[test]
2357 fn legacy_still_executing_consumes_its_status_message() {
2358 let mut data = vec![R_STILL_EXECUTING as u8];
2359 data.extend_from_slice(&7i32.to_le_bytes());
2360 data.extend_from_slice(b"running");
2361
2362 let resp = parse_legacy_response(&data).unwrap();
2363 assert!(matches!(resp.terminal, NativeResponse::StillExecuting));
2364 }
2365
2366 #[test]
2367 fn legacy_still_executing_with_a_short_status_message_is_rejected() {
2368 let mut data = vec![R_STILL_EXECUTING as u8];
2369 data.extend_from_slice(&10i32.to_le_bytes());
2370 data.extend_from_slice(b"ab");
2371
2372 assert_eq!(
2373 legacy_error(&data),
2374 "StillExecuting status message truncated"
2375 );
2376 }
2377
2378 #[test]
2379 fn legacy_more_rows_takes_the_rest_of_the_buffer() {
2380 let resp = parse_legacy_response(&more_rows_part(&[0xAA, 0xBB, 0xCC])).unwrap();
2381
2382 match resp.terminal {
2383 NativeResponse::MoreRows(payload) => assert_eq!(payload, vec![0xAA, 0xBB, 0xCC]),
2384 other => panic!("Expected MoreRows, got {other:?}"),
2385 }
2386 }
2387
2388 #[test]
2389 fn legacy_unknown_response_type_is_rejected() {
2390 assert_eq!(legacy_error(&[0x7F]), "Unknown response type: 127");
2391 }
2392
2393 #[test]
2394 fn legacy_response_with_unconsumed_bytes_is_rejected() {
2395 let mut data = empty_part();
2396 data.push(0xAA);
2397
2398 assert!(
2399 legacy_error(&data).starts_with("Trailing bytes after legacy response:"),
2400 "unexpected message"
2401 );
2402 }
2403
2404 #[test]
2407 fn handle_sub_result_with_small_resultset_handle_describes_result_columns() {
2408 let sub = result_set_part(build_result_set_header_with_handle(
2409 SMALL_RESULTSET,
2410 &[("C1", T_INTEGER, Vec::new())],
2411 ));
2412
2413 let resp = parse_legacy_response(&handle_part(5, &[sub])).unwrap();
2414
2415 match resp.terminal {
2416 NativeResponse::PreparedStatement {
2417 handle,
2418 parameters,
2419 result_columns,
2420 } => {
2421 assert_eq!(handle, 5);
2422 assert!(parameters.is_empty());
2423 assert_eq!(result_columns.len(), 1);
2424 assert_eq!(result_columns[0].name, "C1");
2425 }
2426 other => panic!("Expected PreparedStatement, got {other:?}"),
2427 }
2428 }
2429
2430 #[test]
2431 fn handle_sub_result_with_parameter_description_handle_describes_parameters() {
2432 let sub = result_set_part(build_result_set_header_with_handle(
2433 PARAMETER_DESCRIPTION,
2434 &[("P1", T_CHAR, varchar_meta_bytes(100))],
2435 ));
2436
2437 let resp = parse_legacy_response(&handle_part(5, &[sub])).unwrap();
2438
2439 match resp.terminal {
2440 NativeResponse::PreparedStatement {
2441 handle,
2442 parameters,
2443 result_columns,
2444 } => {
2445 assert_eq!(handle, 5);
2446 assert_eq!(parameters.len(), 1);
2447 assert_eq!(parameters[0].name, "P1");
2448 assert!(parameters[0].is_varchar);
2449 assert!(result_columns.is_empty());
2450 }
2451 other => panic!("Expected PreparedStatement, got {other:?}"),
2452 }
2453 }
2454
2455 #[test]
2456 fn handle_with_both_sub_results_keeps_parameters_and_result_columns() {
2457 let first_result_set = result_set_part(build_result_set_header_with_handle(
2458 SMALL_RESULTSET,
2459 &[("C1", T_INTEGER, Vec::new())],
2460 ));
2461 let parameter_description = result_set_part(build_result_set_header_with_handle(
2462 PARAMETER_DESCRIPTION,
2463 &[("P1", T_INTEGER, Vec::new())],
2464 ));
2465 let last_result_set = result_set_part(build_result_set_header_with_handle(
2466 7,
2467 &[("C2", T_INTEGER, Vec::new())],
2468 ));
2469
2470 let resp = parse_legacy_response(&handle_part(
2471 5,
2472 &[first_result_set, parameter_description, last_result_set],
2473 ))
2474 .unwrap();
2475
2476 match resp.terminal {
2477 NativeResponse::PreparedStatement {
2478 handle,
2479 parameters,
2480 result_columns,
2481 } => {
2482 assert_eq!(handle, 5);
2483 assert_eq!(parameters.len(), 1);
2484 assert_eq!(parameters[0].name, "P1");
2485 assert_eq!(result_columns.len(), 1);
2486 assert_eq!(result_columns[0].name, "C2");
2487 }
2488 other => panic!("Expected PreparedStatement, got {other:?}"),
2489 }
2490 }
2491
2492 #[test]
2493 fn handle_sub_result_that_is_not_a_result_set_leaves_no_columns() {
2494 let resp = parse_legacy_response(&handle_part(5, &[empty_part()])).unwrap();
2495
2496 match resp.terminal {
2497 NativeResponse::PreparedStatement {
2498 handle,
2499 parameters,
2500 result_columns,
2501 } => {
2502 assert_eq!(handle, 5);
2503 assert!(parameters.is_empty());
2504 assert!(result_columns.is_empty());
2505 }
2506 other => panic!("Expected PreparedStatement, got {other:?}"),
2507 }
2508 }
2509
2510 #[test]
2511 fn handle_sub_result_carrying_an_exception_is_rejected() {
2512 let data = handle_part(5, &[exception_part("prepare refused", "42000")]);
2513
2514 let msg = legacy_error(&data);
2515
2516 assert!(msg.contains("prepare refused"), "{msg}");
2517 assert!(msg.contains("42000"), "{msg}");
2518 }
2519
2520 #[test]
2521 fn handle_sub_result_warning_is_propagated() {
2522 let data = handle_part(5, &[warning_part("truncated", "01004")]);
2523
2524 let resp = parse_legacy_response(&data).unwrap();
2525
2526 assert_eq!(resp.warnings.len(), 1);
2527 assert_eq!(resp.warnings[0].message, "truncated");
2528 assert_eq!(resp.warnings[0].sql_state, "01004");
2529 match resp.terminal {
2530 NativeResponse::PreparedStatement {
2531 handle,
2532 parameters,
2533 result_columns,
2534 } => {
2535 assert_eq!(handle, 5);
2536 assert!(parameters.is_empty());
2537 assert!(result_columns.is_empty());
2538 }
2539 other => panic!("Expected PreparedStatement, got {other:?}"),
2540 }
2541 }
2542
2543 #[test]
2544 fn counted_envelope_handle_part_propagates_a_sub_result_warning() {
2545 let data = envelope(&[handle_part(31, &[warning_part("careful", "01000")])]);
2546
2547 let resp = parse_response(&data).unwrap();
2548
2549 assert_eq!(resp.warnings.len(), 1);
2550 assert_eq!(resp.warnings[0].message, "careful");
2551 assert!(matches!(
2552 resp.terminal,
2553 NativeResponse::PreparedStatement { handle: 31, .. }
2554 ));
2555 }
2556
2557 #[test]
2560 fn more_rows_as_the_final_envelope_part_is_accepted() {
2561 let data = envelope(&[column_count_part(2), more_rows_part(&[1, 2, 3])]);
2562
2563 let resp = parse_response(&data).unwrap();
2564
2565 match resp.terminal {
2566 NativeResponse::MoreRows(payload) => assert_eq!(payload, vec![1, 2, 3]),
2567 other => panic!("Expected MoreRows, got {other:?}"),
2568 }
2569 }
2570
2571 #[test]
2572 fn more_rows_before_the_final_envelope_part_is_rejected() {
2573 let data = envelope(&[more_rows_part(&[1, 2, 3]), empty_part()]);
2574
2575 assert_eq!(
2576 parse_error(&data),
2577 "MoreRows must be the final native response part"
2578 );
2579 }
2580
2581 #[test]
2582 fn two_terminal_results_in_one_envelope_are_rejected() {
2583 let data = envelope(&[row_count_part(1), row_count_part(2)]);
2584
2585 assert_eq!(
2586 parse_error(&data),
2587 "Multiple terminal results in native response envelope"
2588 );
2589 }
2590
2591 #[test]
2592 fn unknown_response_type_inside_an_envelope_names_its_offset() {
2593 let data = envelope(&[column_count_part(1), vec![0x7F]]);
2594
2595 let msg = parse_error(&data);
2596 assert!(
2597 msg.starts_with("Unknown response type: 127 at offset 9"),
2598 "unexpected message: {msg}"
2599 );
2600 }
2601
2602 #[test]
2603 fn envelope_with_an_exception_part_returns_the_exception() {
2604 let data = envelope(&[
2605 warning_part("careful", "01000"),
2606 exception_part("bad", "42000"),
2607 ]);
2608
2609 let resp = parse_response(&data).unwrap();
2610
2611 assert_eq!(resp.warnings.len(), 1);
2612 match resp.terminal {
2613 NativeResponse::Exception { message, sql_state } => {
2614 assert_eq!(message, "bad");
2615 assert_eq!(sql_state, "42000");
2616 }
2617 other => panic!("Expected Exception, got {other:?}"),
2618 }
2619 }
2620
2621 #[test]
2622 fn envelope_with_a_handle_part_returns_the_statement_handle() {
2623 let data = envelope(&[handle_part(31, &[])]);
2624
2625 let resp = parse_response(&data).unwrap();
2626
2627 match resp.terminal {
2628 NativeResponse::PreparedStatement {
2629 handle,
2630 parameters,
2631 result_columns,
2632 } => {
2633 assert_eq!(handle, 31);
2634 assert!(parameters.is_empty());
2635 assert!(result_columns.is_empty());
2636 }
2637 other => panic!("Expected PreparedStatement, got {other:?}"),
2638 }
2639 }
2640
2641 #[test]
2645 fn payload_too_short_for_a_result_count_falls_back_to_legacy() {
2646 let resp = parse_response(&[R_STILL_EXECUTING as u8]).unwrap();
2647 assert!(matches!(resp.terminal, NativeResponse::StillExecuting));
2648 }
2649
2650 #[test]
2653 fn out_of_range_result_count_falls_back_to_legacy() {
2654 let data = row_count_part(42);
2655 assert_eq!(
2656 i32::from_le_bytes([data[0], data[1], data[2], data[3]]),
2657 10_752
2658 );
2659
2660 let resp = parse_response(&data).unwrap();
2661 assert!(matches!(resp.terminal, NativeResponse::RowCount(42)));
2662 }
2663
2664 #[test]
2667 fn unknown_first_part_type_falls_back_to_legacy() {
2668 let handle = i32::from_le_bytes([0x00, 0x00, 0x00, 0x63]);
2669 let data = result_set_part(build_result_set_header_with_handle(handle, &[]));
2670 assert_eq!(i32::from_le_bytes([data[0], data[1], data[2], data[3]]), 1);
2671 assert!(!is_known_response_type(data[4] as i8));
2672
2673 let resp = parse_response(&data).unwrap();
2674
2675 match resp.terminal {
2676 NativeResponse::ResultSet {
2677 handle: parsed,
2678 batch,
2679 ..
2680 } => {
2681 assert_eq!(parsed, handle);
2682 assert!(batch.is_none());
2683 }
2684 other => panic!("Expected ResultSet, got {other:?}"),
2685 }
2686 }
2687
2688 #[test]
2691 fn counted_envelope_leaving_trailing_bytes_falls_back_to_legacy() {
2692 let mut data = 6i32.to_le_bytes().to_vec();
2693 for _ in 0..6 {
2694 data.extend_from_slice(&column_count_part(0));
2695 }
2696 data.push(0xAA);
2697
2698 let resp = parse_response(&data).unwrap();
2699
2700 match resp.terminal {
2701 NativeResponse::MoreRows(payload) => assert_eq!(payload, data[1..].to_vec()),
2702 other => panic!("Expected MoreRows, got {other:?}"),
2703 }
2704 }
2705
2706 #[test]
2707 fn known_response_types_are_exactly_the_nine_protocol_parts() {
2708 for result_type in [
2709 R_ROW_COUNT,
2710 R_RESULT_SET,
2711 R_HANDLE,
2712 R_COLUMN_COUNT,
2713 R_WARNING,
2714 R_STILL_EXECUTING,
2715 R_MORE_ROWS,
2716 R_EXCEPTION,
2717 R_EMPTY,
2718 ] {
2719 assert!(
2720 is_known_response_type(result_type),
2721 "{result_type} should be known"
2722 );
2723 }
2724 for result_type in [7i8, 99, -3, -128] {
2725 assert!(
2726 !is_known_response_type(result_type),
2727 "{result_type} should be unknown"
2728 );
2729 }
2730 }
2731
2732 #[test]
2735 fn message_part_shorter_than_its_length_prefix_is_rejected() {
2736 let mut data = vec![R_EXCEPTION as u8];
2737 data.extend_from_slice(&10i32.to_le_bytes());
2738 data.extend_from_slice(b"ab");
2739
2740 assert_eq!(legacy_error(&data), "Exception message truncated");
2741 }
2742
2743 #[test]
2744 fn non_utf8_message_part_is_rejected() {
2745 let mut data = vec![R_EXCEPTION as u8];
2746 data.extend_from_slice(&2i32.to_le_bytes());
2747 data.extend_from_slice(&[0xFF, 0xFE]);
2748 data.extend_from_slice(b"42000");
2749
2750 assert!(
2751 legacy_error(&data).starts_with("Invalid exception UTF-8: "),
2752 "unexpected message"
2753 );
2754 }
2755
2756 #[test]
2757 fn message_part_without_room_for_a_sql_state_uses_the_unknown_state() {
2758 let mut data = 2i32.to_le_bytes().to_vec();
2759 data.extend_from_slice(b"ab");
2760 data.extend_from_slice(b"123");
2761
2762 let mut offset = 0;
2763 let warning = parse_message_part(&data, &mut offset).unwrap();
2764
2765 assert_eq!(warning.message, "ab");
2766 assert_eq!(warning.sql_state, "?????");
2767 assert_eq!(offset, 6);
2768 }
2769
2770 #[test]
2771 fn message_part_with_a_non_utf8_sql_state_uses_the_unknown_state() {
2772 let mut data = 2i32.to_le_bytes().to_vec();
2773 data.extend_from_slice(b"ab");
2774 data.extend_from_slice(&[0xFFu8; 5]);
2775
2776 let mut offset = 0;
2777 let warning = parse_message_part(&data, &mut offset).unwrap();
2778
2779 assert_eq!(warning.sql_state, "?????");
2780 assert_eq!(offset, data.len());
2781 }
2782
2783 #[test]
2786 fn column_name_longer_than_the_buffer_is_rejected() {
2787 let mut data = SMALL_RESULTSET.to_le_bytes().to_vec();
2788 data.extend_from_slice(&1i32.to_le_bytes());
2789 data.extend_from_slice(&0i64.to_le_bytes());
2790 data.extend_from_slice(&0i64.to_le_bytes());
2791 data.extend_from_slice(&50i32.to_le_bytes());
2792 data.extend_from_slice(b"ab");
2793
2794 assert_eq!(
2795 protocol_message(parse_result_set(&data).unwrap_err()),
2796 "Column name truncated"
2797 );
2798 }
2799
2800 #[test]
2801 fn non_utf8_column_name_is_rejected() {
2802 let mut data = SMALL_RESULTSET.to_le_bytes().to_vec();
2803 data.extend_from_slice(&1i32.to_le_bytes());
2804 data.extend_from_slice(&0i64.to_le_bytes());
2805 data.extend_from_slice(&0i64.to_le_bytes());
2806 data.extend_from_slice(&2i32.to_le_bytes());
2807 data.extend_from_slice(&[0xFF, 0xFE]);
2808 data.extend_from_slice(&T_INTEGER.to_le_bytes());
2809
2810 assert!(
2811 protocol_message(parse_result_set(&data).unwrap_err())
2812 .starts_with("Invalid column name UTF-8: "),
2813 "unexpected message"
2814 );
2815 }
2816
2817 #[test]
2818 fn string_like_column_metadata_without_its_varchar_flag_is_rejected() {
2819 for (type_id, expected) in [
2820 (T_CHAR, "CHAR metadata truncated"),
2821 (T_GEOMETRY, "CHAR metadata truncated"),
2822 (T_HASHTYPE, "CHAR metadata truncated"),
2823 (T_INTERVAL_YEAR, "INTERVAL metadata truncated"),
2824 (T_INTERVAL_DAY, "INTERVAL metadata truncated"),
2825 ] {
2826 let data = build_result_set_header(&[("C", type_id, Vec::new())]);
2827
2828 assert_eq!(
2829 protocol_message(parse_result_set(&data).unwrap_err()),
2830 expected,
2831 "type_id {type_id}"
2832 );
2833 }
2834 }
2835
2836 #[test]
2837 fn negative_rows_received_in_a_result_set_is_rejected() {
2838 let mut data = SMALL_RESULTSET.to_le_bytes().to_vec();
2839 data.extend_from_slice(&0i32.to_le_bytes());
2840 data.extend_from_slice(&0i64.to_le_bytes());
2841 data.extend_from_slice(&(-1i64).to_le_bytes());
2842
2843 assert_eq!(
2844 protocol_message(parse_result_set(&data).unwrap_err()),
2845 "Invalid row count from server: -1"
2846 );
2847 }
2848
2849 #[test]
2850 fn negative_rows_received_in_a_fetch_is_rejected() {
2851 let data = (-5i64).to_le_bytes();
2852
2853 assert_eq!(
2854 protocol_message(parse_fetch_to_record_batch(&data, &[]).unwrap_err()),
2855 "Invalid row count from server: -5"
2856 );
2857 }
2858
2859 #[test]
2860 fn batch_without_columns_still_reports_its_row_count() {
2861 let data = Vec::new();
2862 let mut offset = 0;
2863
2864 let batch = build_batch_from_wire(&data, &mut offset, &[], 3).unwrap();
2865
2866 assert_eq!(batch.num_rows(), 3);
2867 assert_eq!(batch.num_columns(), 0);
2868 assert_eq!(offset, 0);
2869 }
2870
2871 #[test]
2874 fn truncated_column_values_name_the_missing_primitive() {
2875 let cases: Vec<(u32, Vec<u8>, &str)> = vec![
2876 (T_INTEGER, vec![], "Data truncated (u8)"),
2877 (T_BOOLEAN, vec![1u8], "Data truncated (u8)"),
2878 (T_DOUBLE, vec![1u8, 0, 0], "Data truncated (f64)"),
2879 (T_REAL, vec![1u8, 0], "Data truncated (f32)"),
2880 (T_INTEGER, vec![1u8, 0, 0, 0], "Data truncated (i64)"),
2881 (T_SMALLINT, vec![1u8, 0], "Data truncated (i32)"),
2882 (T_DATE, vec![1u8, 0], "Data truncated (i32)"),
2883 (T_BIGDECIMAL, vec![1u8, 0, 0], "Data truncated (i128)"),
2884 (T_TIMESTAMP, vec![1u8, 0], "Data truncated (i16)"),
2885 ];
2886
2887 for (type_id, data, expected) in cases {
2888 let columns = vec![meta("c", type_id, None, None)];
2889 let mut offset = 0;
2890
2891 let err = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2892
2893 assert_eq!(protocol_message(err), expected, "type_id {type_id}");
2894 }
2895 }
2896
2897 #[test]
2898 fn string_value_longer_than_the_buffer_is_rejected() {
2899 let columns = vec![meta("s", T_CHAR, None, None)];
2900 let mut data = vec![1u8];
2901 data.extend_from_slice(&10i32.to_le_bytes());
2902 data.extend_from_slice(b"ab");
2903
2904 let mut offset = 0;
2905 let err = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2906
2907 assert_eq!(protocol_message(err), "String data truncated");
2908 }
2909
2910 #[test]
2911 fn non_utf8_string_value_is_rejected() {
2912 let columns = vec![meta("s", T_CHAR, None, None)];
2913 let mut data = vec![1u8];
2914 data.extend_from_slice(&2i32.to_le_bytes());
2915 data.extend_from_slice(&[0xFF, 0xFE]);
2916
2917 let mut offset = 0;
2918 let err = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2919
2920 assert!(
2921 protocol_message(err).starts_with("Invalid UTF-8 in string: "),
2922 "unexpected message"
2923 );
2924 }
2925
2926 #[test]
2927 fn binary_value_longer_than_the_buffer_is_rejected() {
2928 let columns = vec![meta("b", T_BINARY, None, None)];
2929 let mut data = vec![1u8];
2930 data.extend_from_slice(&10i32.to_le_bytes());
2931 data.extend_from_slice(b"ab");
2932
2933 let mut offset = 0;
2934 let err = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2935
2936 assert_eq!(protocol_message(err), "Binary data truncated");
2937 }
2938
2939 #[test]
2941 fn interval_value_longer_than_the_buffer_is_rejected() {
2942 let columns = vec![meta("IV", T_INTERVAL_DAY, None, None)];
2943 let mut data = vec![1u8];
2944 data.extend_from_slice(&10i32.to_le_bytes());
2945 data.extend_from_slice(b"ab");
2946
2947 let mut offset = 0;
2948 let err = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2949
2950 assert_eq!(protocol_message(err), "Interval data truncated");
2951 }
2952
2953 #[test]
2955 fn non_utf8_interval_value_is_rejected() {
2956 let columns = vec![meta("IV", T_INTERVAL_DAY, None, None)];
2957 let mut data = vec![1u8];
2958 data.extend_from_slice(&2i32.to_le_bytes());
2959 data.extend_from_slice(&[0xFF, 0xFE]);
2960
2961 let mut offset = 0;
2962 let err = build_batch_from_wire(&data, &mut offset, &columns, 1).unwrap_err();
2963
2964 assert!(
2965 protocol_message(err).starts_with("Invalid UTF-8 in INTERVAL column IV: "),
2966 "unexpected message"
2967 );
2968 }
2969
2970 #[test]
2971 fn decimal_columns_accept_nulls_on_both_wire_widths() {
2972 use arrow::array::{Array, Decimal128Array};
2973
2974 for (precision, value_bytes) in [
2975 (9, 1_234i32.to_le_bytes().to_vec()),
2976 (18, 1_234i64.to_le_bytes().to_vec()),
2977 ] {
2978 let columns = vec![meta("d", T_DECIMAL, Some(precision), Some(0))];
2979 let mut data = vec![1u8];
2980 data.extend_from_slice(&value_bytes);
2981 data.push(0u8);
2982
2983 let mut offset = 0;
2984 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
2985
2986 let arr = batch
2987 .column(0)
2988 .as_any()
2989 .downcast_ref::<Decimal128Array>()
2990 .unwrap();
2991 assert_eq!(arr.value(0), 1_234i128, "precision {precision}");
2992 assert!(arr.is_null(1), "precision {precision}");
2993 assert_eq!(offset, data.len(), "precision {precision}");
2994 }
2995 }
2996
2997 #[test]
2998 fn bigdecimal_column_accepts_nulls() {
2999 use arrow::array::{Array, Decimal128Array};
3000
3001 let columns = vec![meta("bd", T_BIGDECIMAL, Some(36), Some(0))];
3002 let mut data = vec![1u8];
3003 data.extend_from_slice(&7i128.to_le_bytes());
3004 data.push(0u8);
3005
3006 let mut offset = 0;
3007 let batch = build_batch_from_wire(&data, &mut offset, &columns, 2).unwrap();
3008
3009 let arr = batch
3010 .column(0)
3011 .as_any()
3012 .downcast_ref::<Decimal128Array>()
3013 .unwrap();
3014 assert_eq!(arr.value(0), 7i128);
3015 assert!(arr.is_null(1));
3016 assert_eq!(offset, data.len());
3017 }
3018
3019 #[test]
3020 fn result_set_with_unconsumed_bytes_is_rejected() {
3021 let mut data = build_result_set_header(&[]);
3022 data.push(0xAA);
3023
3024 assert!(
3025 protocol_message(parse_result_set(&data).unwrap_err())
3026 .starts_with("Trailing bytes after result-set response:"),
3027 "unexpected message"
3028 );
3029 }
3030
3031 #[test]
3032 fn result_set_promising_rows_it_does_not_carry_is_rejected() {
3033 let mut data = SMALL_RESULTSET.to_le_bytes().to_vec();
3034 data.extend_from_slice(&1i32.to_le_bytes());
3035 data.extend_from_slice(&2i64.to_le_bytes());
3036 data.extend_from_slice(&2i64.to_le_bytes());
3037 write_col_meta(&mut data, "I", T_INTEGER, &[]);
3038
3039 assert_eq!(
3040 protocol_message(parse_result_set(&data).unwrap_err()),
3041 "Data truncated (u8)"
3042 );
3043 }
3044
3045 #[test]
3048 fn decimal_precision_and_scale_are_clamped_to_the_arrow_range() {
3049 let cases = [
3050 (None, None, 18, (18u8, 0i8)),
3051 (Some(100), Some(2), 18, (38, 2)),
3052 (Some(0), Some(2), 18, (1, 2)),
3053 (Some(10), Some(1_000), 18, (10, 127)),
3054 (Some(10), Some(-1_000), 18, (10, -128)),
3055 ];
3056
3057 for (precision, scale, default_precision, expected) in cases {
3058 let column = meta("d", T_DECIMAL, precision, scale);
3059
3060 assert_eq!(
3061 decimal_precision_scale(&column, default_precision),
3062 expected,
3063 "precision {precision:?}, scale {scale:?}"
3064 );
3065 }
3066 }
3067}