Skip to main content

exarrow_rs/transport/native/
result_parser.rs

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
21/// One decoded column: the Arrow field describing it and the array holding its values.
22type BuiltColumn = (Field, ArrayRef);
23
24/// Parsed column metadata from a native protocol result set.
25#[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    /// Declared fractional second precision of an INTERVAL DAY TO SECOND column.
34    pub interval_fraction: Option<i32>,
35}
36
37/// Parsed response from the native protocol.
38#[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    /// Reply to `CMD_CREATE_PREPARED`: two independent column descriptions and no
48    /// row data. This is a shape of its own because `ResultSet` holds a single
49    /// column list, which forced one of the two descriptions to be discarded.
50    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/// Non-fatal warning returned alongside a terminal result.
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct NativeWarning {
68    pub message: String,
69    pub sql_state: String,
70}
71
72/// Parsed native response envelope containing warnings and one terminal result.
73#[derive(Debug)]
74pub struct NativeResponseEnvelope {
75    pub warnings: Vec<NativeWarning>,
76    pub terminal: NativeResponse,
77}
78
79/// Parse a response payload (after the 21-byte header) into a structured response.
80pub 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
306/// Parse a handle-only response (R_HANDLE = 2), sent in reply to CREATE PREPARED.
307///
308/// Format: [statement_handle:4 LE] [sub_result...]
309///
310/// A result-set sub-result whose handle is `PARAMETER_DESCRIPTION` describes the
311/// parameters; any other describes the statement's result-set columns. An
312/// exception sub-result is reported as an error rather than as a handle with
313/// empty metadata, which the caller would otherwise bind against as though the
314/// server had confirmed it.
315fn 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            // All string-like types: vcFlag(1) + maxLen(4) + octetLen(4)
440            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            // octet_len
450            let _octet_len = read_i32(data, offset)?;
451        }
452        T_INTERVAL_YEAR => {
453            // vcFlag(1) + maxLen(4) + octetLen(4)
454            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            // Protocol v19+: datetimeIntervalPrecision(4)
465            let _interval_precision = read_i32(data, offset)?;
466        }
467        T_INTERVAL_DAY => {
468            // vcFlag(1) + maxLen(4) + octetLen(4)
469            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            // Protocol v19+: datetimeIntervalPrecision(4) + datetimeIntervalFraction(4)
480            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            // Protocol v19+: precision(4)
485            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
544/// Build an Arrow `RecordBatch` directly from native wire bytes in a single pass,
545/// appending values straight into typed Arrow builders. DATE → Date32,
546/// TIMESTAMP → TimestampMicrosecond; no intermediate string allocations.
547fn 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
577/// Extract Arrow Decimal128 precision and scale from column metadata, clamping
578/// to the valid Arrow range (precision 1..=38, scale in i8) and defaulting
579/// precision when absent.
580fn 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
588/// Parse one column worth of wire bytes directly into an Arrow array.
589///
590/// Types that Exasol sends as length-prefixed text convert to the Arrow type of
591/// `TypeMapper::exasol_to_arrow`: both INTERVAL kinds parse into Interval(MonthDayNano),
592/// and GEOMETRY and HASHTYPE keep their text bytes in Binary. The `T_INTEGER` and
593/// `T_SMALLINT` wire types decode to the Decimal128 type of the DECIMAL(18,0) and DECIMAL(9,0)
594/// that `native_meta_to_data_type` reports, and the wire type alone fixes the 8-byte or 4-byte
595/// value width and that Decimal128 type, whatever precision the column description carries.
596/// CHAR and any type id this driver does not recognise decode as UTF-8 strings.
597fn 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
886/// Decode the packed `[year:16][month:8][day:8]` DATE value into Arrow Date32 days.
887fn 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
908/// Read a length-prefixed value, reporting `truncated_message` when the declared
909/// length runs past the end of the buffer.
910fn 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
924/// Specialised length-prefixed string path that appends directly to a
925/// `StringBuilder` without allocating an owned `String` per value.
926fn 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
966// --- Primitive readers ---
967
968fn 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    /// Helper: write column metadata bytes for a given type.
1143    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    /// Helper: build a minimal result set payload (after the result-type byte).
1194    /// Contains only column metadata (0 rows) so no column data section.
1195    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        // total_rows
1207        data.extend_from_slice(&0i64.to_le_bytes());
1208        // rows_received
1209        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    /// Metadata bytes a CHAR column carries: vcFlag(1) + maxLen(4) + octetLen(4).
1240    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        // TIMESTAMP metadata in protocol v19+: type_id(4) + precision(4)
1309        let mut ts_extra = Vec::new();
1310        ts_extra.extend_from_slice(&3i32.to_le_bytes()); // precision = 3
1311
1312        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()); // precision = 6
1330
1331        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        // INTERVAL YEAR TO MONTH: vcFlag(1) + maxLen(4) + octetLen(4) + intervalPrecision(4)
1348        let mut extra = Vec::new();
1349        extra.push(0u8); // vcFlag
1350        extra.extend_from_slice(&13i32.to_le_bytes()); // maxLen
1351        extra.extend_from_slice(&13i32.to_le_bytes()); // octetLen
1352        extra.extend_from_slice(&2i32.to_le_bytes()); // intervalPrecision
1353
1354        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        // INTERVAL DAY TO SECOND: vcFlag(1) + maxLen(4) + octetLen(4)
1372        //   + intervalPrecision(4) + intervalFraction(4)
1373        let mut extra = Vec::new();
1374        extra.push(0u8); // vcFlag
1375        extra.extend_from_slice(&29i32.to_le_bytes()); // maxLen
1376        extra.extend_from_slice(&29i32.to_le_bytes()); // octetLen
1377        extra.extend_from_slice(&2i32.to_le_bytes()); // intervalPrecision
1378        extra.extend_from_slice(&6i32.to_le_bytes()); // intervalFraction
1379
1380        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        // GEOMETRY: vcFlag(1) + maxLen(4) + octetLen(4)
1398        let mut extra = Vec::new();
1399        extra.push(0u8); // vcFlag (no varchar, no UTF-8)
1400        extra.extend_from_slice(&100i32.to_le_bytes()); // maxLen
1401        extra.extend_from_slice(&100i32.to_le_bytes()); // octetLen
1402
1403        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        // Literal vcFlag bytes, not the constants the parser reads them with:
1420        // building the input from IS_VARCHAR would pass for any mask value.
1421        let reported_type = |vc_flag: u8| {
1422            let mut extra = vec![vc_flag];
1423            extra.extend_from_slice(&50i32.to_le_bytes()); // maxLen
1424            extra.extend_from_slice(&200i32.to_le_bytes()); // octetLen
1425            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        // Simulate the benchmark table:
1445        // id BIGINT, name VARCHAR(100), email VARCHAR(200), age INTEGER,
1446        // salary DECIMAL(12,2), created_at TIMESTAMP, is_active BOOLEAN,
1447        // description VARCHAR(1000)
1448
1449        let mut columns = Vec::new();
1450
1451        // BIGINT -> DECIMAL(18,0) on the wire
1452        let mut decimal_extra = Vec::new();
1453        decimal_extra.extend_from_slice(&18i32.to_le_bytes()); // precision
1454        decimal_extra.extend_from_slice(&0i32.to_le_bytes()); // scale
1455        columns.push(("ID", T_DECIMAL, decimal_extra));
1456
1457        // VARCHAR(100)
1458        let mut vc_extra = Vec::new();
1459        vc_extra.push(IS_VARCHAR | IS_UTF8); // vcFlag
1460        vc_extra.extend_from_slice(&100i32.to_le_bytes()); // maxLen
1461        vc_extra.extend_from_slice(&400i32.to_le_bytes()); // octetLen (UTF-8 = 4x)
1462        columns.push(("NAME", T_CHAR, vc_extra));
1463
1464        // VARCHAR(200)
1465        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        // INTEGER -> DECIMAL(18,0) on the wire
1472        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        // DECIMAL(12,2)
1478        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        // TIMESTAMP with precision field (protocol v19+)
1484        let mut ts_extra = Vec::new();
1485        ts_extra.extend_from_slice(&3i32.to_le_bytes()); // precision
1486        columns.push(("CREATED_AT", T_TIMESTAMP, ts_extra));
1487
1488        // BOOLEAN (no extra metadata)
1489        columns.push(("IS_ACTIVE", T_BOOLEAN, Vec::new()));
1490
1491        // VARCHAR(1000)
1492        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        // Row 0: 2024-01-02 → packed i32
1556        let packed = (2024i32 << 16) | (1 << 8) | 2;
1557        data.push(1u8);
1558        data.extend_from_slice(&packed.to_le_bytes());
1559        // Row 1: null
1560        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    /// Scenario: Pre-1970 DATE and TIMESTAMP query results keep their calendar day
1578    #[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        // Row 0: 1970-01-01 00:00:00.123
1627        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()); // 123 ms → 123_000 micros
1635                                                               // Row 1: null
1636        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        // Row 0: "hello"
1681        data.push(1u8);
1682        data.extend_from_slice(&5i32.to_le_bytes());
1683        data.extend_from_slice(b"hello");
1684        // Row 1: null
1685        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        // DOUBLE + BOOLEAN + INTEGER with 2 rows each, mixed nulls.
1701        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        // DOUBLE column
1708        data.push(1u8);
1709        data.extend_from_slice(&1.5f64.to_le_bytes());
1710        data.push(0u8); // null
1711
1712        // BOOLEAN column
1713        data.push(1u8);
1714        data.push(1u8); // true
1715        data.push(0u8); // null
1716
1717        // INTEGER column
1718        data.push(1u8);
1719        data.extend_from_slice(&42i64.to_le_bytes());
1720        data.push(0u8); // null
1721
1722        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        // rows_received prefix
1799        data.extend_from_slice(&2i64.to_le_bytes());
1800        // Row 0: 100
1801        data.push(1u8);
1802        data.extend_from_slice(&100i64.to_le_bytes());
1803        // Row 1: null
1804        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        // precision <= 9 would be i32; precision 12 uses i64
1849        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    /// Scenario: Direct binary to Arrow conversion for numeric types
1868    #[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); // null
1878
1879        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    /// Scenario: Direct binary to Arrow conversion for numeric types
1945    #[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    /// Scenario: Direct binary to Arrow conversion for numeric types
1977    #[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    /// Scenario: Direct binary to Arrow conversion for numeric types
2008    #[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    /// Scenario: Direct binary to Arrow conversion for numeric types
2030    #[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    /// Scenario: Direct binary to Arrow conversion for interval types
2142    #[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    /// Scenario: Direct binary to Arrow conversion for interval types
2182    #[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    /// Scenario: Direct binary to Arrow conversion for interval types
2196    #[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    /// Scenario: Direct binary to Arrow conversion for binary and geometry types
2208    #[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    /// Scenario: Direct binary to Arrow conversion for numeric types
2237    #[test]
2238    fn single_pass_bigdecimal_preserves_i128() {
2239        use arrow::array::Decimal128Array;
2240        use arrow::datatypes::DataType;
2241
2242        // i128 value larger than i64::MAX would overflow under the old path.
2243        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    // --- Legacy (un-counted) framing ---
2265
2266    #[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    // --- Handle-only responses and their sub-results ---
2405
2406    #[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    // --- Counted envelope framing ---
2558
2559    #[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    // --- Falling back from the counted envelope to legacy framing ---
2642
2643    /// A payload under five bytes cannot hold a result count, so it is read as legacy.
2644    #[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    /// A legacy row count whose first four bytes read as an out-of-range result
2651    /// count (10752) is not mistaken for a counted envelope.
2652    #[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    /// A legacy result set whose fifth byte is not a known part type is not
2665    /// mistaken for a counted envelope holding one part.
2666    #[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    /// A counted parse that leaves bytes unconsumed is discarded, and the same
2689    /// payload is re-read as a legacy MoreRows response.
2690    #[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    // --- Message parts ---
2733
2734    #[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    // --- Column metadata errors ---
2784
2785    #[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    // --- Truncated column values ---
2872
2873    #[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    /// Scenario: Direct binary to Arrow conversion for interval types
2940    #[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    /// Scenario: Direct binary to Arrow conversion for interval types
2954    #[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    // --- Decimal precision and scale clamping ---
3046
3047    #[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}