uqa_sql/assignment/conversion/
production.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Assignment conversion preserves admitted inputs while their replacements are produced.
8
9use super::{
10    array_scalar_type, column_type_name, validate_vector_dimensions, ColumnType, SQLError,
11};
12use crate::expr::{
13    cast_value_from_with_control, parse_pg_array_literal_with_control,
14    value_to_tensor_with_control, value_to_text_with_control, value_to_vector_with_control,
15};
16use uqa_core::{
17    memory::{MemoryReservation, Produced, ProductionControl, ProductionString, ProductionVec},
18    ArrayValue, DecimalValue, Value,
19};
20
21type Result<T> = std::result::Result<T, SQLError>;
22
23/// Preserve already admitted carriers without repeating their input function; new text input still follows the selected conversion.
24pub fn convert_value_to_column_type_from_with_control(
25    value: Produced<Value>,
26    ty: &ColumnType,
27    source: Option<&ColumnType>,
28    control: &ProductionControl<'_>,
29) -> Result<Produced<Value>> {
30    control.check()?;
31    let (value, memory) = value.into_parts();
32    let value = control.finish(value, memory)?;
33    if source.is_some_and(|source| super::same_catalog_value_type(source, ty)) {
34        return super::normalize_legacy_vector_carrier_with_control(&value, ty, control)
35            .map(|normalized| normalized.unwrap_or(value));
36    }
37    if let (ColumnType::InternalChar, Some(source)) = (ty, source) {
38        let name = source.catalog_name_with_control(control)?;
39        return cast_value_from_with_control(&value, "\"char\"", Some(&name), control);
40    }
41    convert_value_to_column_type_with_control(value, ty, control)
42}
43
44#[expect(
45    clippy::too_many_lines,
46    reason = "assignment matrix preserves conversion and validation order"
47)]
48pub fn convert_value_to_column_type_with_control(
49    value: Produced<Value>,
50    ty: &ColumnType,
51    control: &ProductionControl<'_>,
52) -> Result<Produced<Value>> {
53    let (value, memory) = value.into_parts();
54    let value = control.finish(value, memory)?;
55    if matches!(&*value, Value::Null) {
56        return Ok(value);
57    }
58    match ty {
59        ColumnType::Named(name) => Err(SQLError::Routine {
60            sqlstate: "42704".into(),
61            message: format!("type \"{name}\" does not exist"),
62        }),
63        // Label input needs the catalog; catalog-aware assignment converts text before this context-free step.
64        ColumnType::Enum(reference) => match &*value {
65            Value::Enum(label) if label.type_oid() == reference.oid => Ok(value),
66            _ => Err(SQLError::Internal(format!(
67                "enum input for type OID {} requires catalog-aware conversion",
68                reference.oid
69            ))),
70        },
71        // Composite input and coercion need the attributes; catalog-aware assignment converts rows and text before this context-free step.
72        ColumnType::Composite(reference) => match &*value {
73            Value::Record(_) => Ok(value),
74            _ => Err(SQLError::Internal(format!(
75                "composite input for type OID {} requires catalog-aware conversion",
76                reference.oid
77            ))),
78        },
79        ColumnType::SmallInteger => cast_value_from_with_control(&value, "smallint", None, control),
80        ColumnType::Integer => cast_value_from_with_control(&value, "integer", None, control),
81        ColumnType::BigInteger => cast_value_from_with_control(&value, "bigint", None, control),
82        // `oidin` reads the text, so its diagnostics name `oid`.
83        ColumnType::Oid => cast_value_from_with_control(&value, "oid", None, control),
84        ColumnType::Xid => {
85            let converted = cast_value_from_with_control(&value, "bigint", None, control)?;
86            let Value::Int(number) = &*converted else {
87                unreachable!("bigint cast returned a non-integer value");
88            };
89            u32::try_from(*number).map_err(|_| {
90                SQLError::TypeMismatch(format!(
91                    "value {number} is out of range for type {}",
92                    column_type_name(ty)
93                ))
94            })?;
95            Ok(converted)
96        }
97        ColumnType::Boolean => match &*value {
98            Value::Bool(_) => Ok(value),
99            Value::Str(text) => {
100                let boolean = parse_boolean_text(text)
101                    .ok_or_else(|| crate::expr::invalid_boolean_input(text))?;
102                Ok(control.finish(Value::Bool(boolean), control.empty_reservation())?)
103            }
104            other => Err(SQLError::TypeMismatch(format!(
105                "cannot cast {other:?} to boolean"
106            ))),
107        },
108        ColumnType::Void => Ok(control.finish(Value::Void, control.empty_reservation())?),
109        ColumnType::Text | ColumnType::RefCursor | ColumnType::Varchar(None) => {
110            text_value(value_to_text_with_control(&value, control)?, false, control)
111        }
112        ColumnType::Name => cast_value_from_with_control(&value, "name", None, control),
113        ColumnType::Uuid => cast_value_from_with_control(&value, "uuid", None, control),
114        ColumnType::Varchar(Some(length)) => character_value(&value, *length, false, control),
115        ColumnType::Bpchar => {
116            text_value(value_to_text_with_control(&value, control)?, true, control)
117        }
118        ColumnType::Character(length) => character_value(&value, *length, true, control),
119        ColumnType::Real => cast_value_from_with_control(&value, "real", None, control),
120        ColumnType::DoublePrecision => {
121            cast_value_from_with_control(&value, "double precision", None, control)
122        }
123        ColumnType::Numeric { precision, scale } => {
124            numeric_value(value, *precision, *scale, control)
125        }
126        ColumnType::Json => cast_value_from_with_control(&value, "json", None, control),
127        ColumnType::JsonB => cast_value_from_with_control(&value, "jsonb", None, control),
128        // Text reaches a bytea column through its input function, `byteain`, which decodes the hex and escape formats; no other type has an assignment cast to bytea.
129        ColumnType::Bytea => match &*value {
130            Value::Bytes(_) => Ok(value),
131            Value::Str(_) | Value::FixedChar(_) => {
132                cast_value_from_with_control(&value, "bytea", Some("text"), control)
133            }
134            other => Err(SQLError::TypeMismatch(format!(
135                "cannot cast {other:?} to bytea"
136            ))),
137        },
138        ColumnType::InternalChar => {
139            let text = value_to_text_with_control(&value, control)?;
140            Ok(crate::expr::internal_char::carrier_with_control(
141                crate::expr::internal_char::input_byte(&text),
142                control,
143            )?)
144        }
145        ColumnType::Regproc
146        | ColumnType::Regprocedure
147        | ColumnType::Regclass
148        | ColumnType::Regcollation
149        | ColumnType::Regnamespace
150        | ColumnType::Regtype
151        | ColumnType::AclItem => {
152            if matches!(&*value, Value::Int(_) | Value::Str(_)) {
153                Ok(value)
154            } else {
155                text_value(value_to_text_with_control(&value, control)?, false, control)
156            }
157        }
158        ColumnType::PgNodeTree => {
159            cast_value_from_with_control(&value, "pg_node_tree", None, control)
160        }
161        ColumnType::Regrole => match &*value {
162            Value::Int(number) => {
163                u32::try_from(*number).map_err(|_| {
164                    SQLError::TypeMismatch(format!(
165                        "value {number} is out of range for type regrole"
166                    ))
167                })?;
168                Ok(value)
169            }
170            Value::Str(_) | Value::FixedChar(_) => Err(SQLError::Internal(
171                "regrole name conversion requires catalog resolution".into(),
172            )),
173            other => Err(SQLError::TypeMismatch(format!(
174                "cannot cast {other:?} to regrole"
175            ))),
176        },
177        ColumnType::Int2Vector | ColumnType::OidVector => {
178            let name = column_type_name(ty);
179            if matches!(&*value, Value::LegacyVector(vector) if vector.kind().type_name() == name) {
180                Ok(value)
181            } else {
182                cast_value_from_with_control(&value, name, None, control)
183            }
184        }
185        ColumnType::AnyArray => {
186            if value.array_view().is_some() {
187                Ok(value)
188            } else {
189                Err(SQLError::TypeMismatch(format!(
190                    "cannot cast {:?} to anyarray",
191                    *value
192                )))
193            }
194        }
195        ColumnType::Record => match &*value {
196            Value::Record(_) => Ok(value),
197            Value::Row(_) => record_value(value, control),
198            other => Err(SQLError::TypeMismatch(format!(
199                "cannot cast {other:?} to record"
200            ))),
201        },
202        ColumnType::Array(element) => array_value(value, element, control),
203        ColumnType::Date
204        | ColumnType::Time
205        | ColumnType::TimePrecision(_)
206        | ColumnType::TimeTz
207        | ColumnType::TimeTzPrecision(_)
208        | ColumnType::Timestamp
209        | ColumnType::TimestampPrecision(_)
210        | ColumnType::TimestampTz
211        | ColumnType::TimestampTzPrecision(_)
212        | ColumnType::Interval
213        | ColumnType::IntervalWithFields { .. } => {
214            let name = ty.sql_name_with_control(control)?;
215            cast_value_from_with_control(&value, &name, None, control)
216        }
217        ColumnType::Range(subtype) => {
218            cast_value_from_with_control(&value, subtype.range_name(), None, control)
219        }
220        ColumnType::Multirange(subtype) => {
221            cast_value_from_with_control(&value, subtype.multirange_name(), None, control)
222        }
223        ColumnType::Vector(dimensions) => {
224            let vector = value_to_vector_with_control(&value, control)?;
225            validate_vector_dimensions(*dimensions, vector.len())?;
226            vector_value(&vector, control)
227        }
228        ColumnType::Tensor(dimensions) => {
229            let tensor = value_to_tensor_with_control(&value, control)?;
230            for vector in &*tensor {
231                control.check()?;
232                validate_vector_dimensions(*dimensions, vector.len())?;
233            }
234            let mut output = ProductionVec::new(*control);
235            output.reserve(tensor.len())?;
236            for vector in &*tensor {
237                output.push_produced(vector_value(vector, control)?)?;
238            }
239            let (values, memory) = output.finish()?.into_parts();
240            Ok(control.finish(Value::List(values), memory)?)
241        }
242        ColumnType::Domain { base, .. } => {
243            convert_value_to_column_type_with_control(value, base, control)
244        }
245    }
246}
247
248fn numeric_value(
249    value: Produced<Value>,
250    precision: Option<u32>,
251    scale: Option<i32>,
252    control: &ProductionControl<'_>,
253) -> Result<Produced<Value>> {
254    let decimal = match &*value {
255        Value::Decimal(_) => {
256            let (Value::Decimal(value), memory) = value.into_parts() else {
257                unreachable!();
258            };
259            control.finish(value, memory)?
260        }
261        Value::Int(number) => DecimalValue::from_i64_with_control(*number, control)?,
262        Value::Bool(boolean) => DecimalValue::from_i64_with_control(i64::from(*boolean), control)?,
263        Value::Float(number) => DecimalValue::from_f64_lossy_with_control(*number, control)?
264            .ok_or_else(|| SQLError::TypeMismatch(format!("cannot cast {number:?} to numeric")))?,
265        Value::Str(text) => DecimalValue::parse_with_control(text, control)?
266            .ok_or_else(|| crate::expr::invalid_numeric_input(text))?,
267        other => {
268            return Err(SQLError::TypeMismatch(format!(
269                "cannot cast {other:?} to numeric"
270            )))
271        }
272    };
273    let rounded = match scale {
274        Some(scale) => decimal
275            .round_to_scale_with_control(scale, control)?
276            .ok_or_else(|| {
277                SQLError::TypeMismatch(format!("cannot round numeric to scale {scale}"))
278            })?,
279        None => decimal,
280    };
281    if let Some(precision) = precision {
282        let scale = scale.unwrap_or(0);
283        if !rounded.fits_precision_with_control(precision, scale, control)? {
284            return Err(numeric_field_overflow(precision, scale));
285        }
286    }
287    let (decimal, memory) = rounded.into_parts();
288    Ok(control.finish(Value::Decimal(decimal), memory)?)
289}
290
291fn text_value(
292    text: Produced<String>,
293    fixed: bool,
294    control: &ProductionControl<'_>,
295) -> Result<Produced<Value>> {
296    let (text, memory) = text.into_parts();
297    Ok(control.finish(
298        if fixed {
299            Value::FixedChar(text)
300        } else {
301            Value::Str(text)
302        },
303        memory,
304    )?)
305}
306
307fn character_value(
308    value: &Value,
309    length: u32,
310    fixed: bool,
311    control: &ProductionControl<'_>,
312) -> Result<Produced<Value>> {
313    let name = if fixed {
314        "character"
315    } else {
316        "character varying"
317    };
318    let length = usize::try_from(length).map_err(|_| {
319        SQLError::TypeMismatch(format!(
320            "{name} length {length} exceeds the platform addressable range"
321        ))
322    })?;
323    let text = value_to_text_with_control(value, control)?;
324    let mut count = 0;
325    let mut end = 0;
326    for (index, character) in text.char_indices() {
327        control.check()?;
328        if count < length {
329            count += 1;
330            end = index + character.len_utf8();
331        } else if character != ' ' {
332            return Err(SQLError::Routine {
333                sqlstate: "22001".into(),
334                message: format!("value too long for type {name}({length})"),
335            });
336        }
337    }
338    let mut output = ProductionString::from_produced(text, *control)?;
339    output.truncate(end)?;
340    if fixed {
341        for _ in count..length {
342            output.push(' ')?;
343        }
344    }
345    text_value(output.finish()?, fixed, control)
346}
347
348fn array_value(
349    value: Produced<Value>,
350    element: &ColumnType,
351    control: &ProductionControl<'_>,
352) -> Result<Produced<Value>> {
353    let array = match &*value {
354        Value::Array(_) => {
355            let (Value::Array(array), memory) = value.into_parts() else {
356                unreachable!();
357            };
358            control.finish(array, memory)?
359        }
360        Value::LegacyVector(_) => {
361            let (Value::LegacyVector(vector), memory) = value.into_parts() else {
362                unreachable!();
363            };
364            control.finish(vector.into_array(), memory)?
365        }
366        Value::List(_) => {
367            let (Value::List(elements), memory) = value.into_parts() else {
368                unreachable!();
369            };
370            ArrayValue::try_new_with_control(control.finish(elements, memory)?, control)?
371                .ok_or_else(array_shape_error)?
372        }
373        Value::Str(text) => parse_pg_array_literal_with_control(text, control)?,
374        other => {
375            return Err(SQLError::TypeMismatch(format!(
376                "cannot cast {other:?} to {}[]",
377                column_type_name(element)
378            )))
379        }
380    };
381    let converted = array_elements(array.elements(), array_scalar_type(element), control)?;
382    let mut bounds = ProductionVec::new(*control);
383    bounds.reserve(array.lower_bounds().len())?;
384    for bound in array.lower_bounds() {
385        bounds.push_copy(*bound)?;
386    }
387    let array = ArrayValue::with_lower_bounds_with_control(converted, bounds.finish()?, control)?
388        .ok_or_else(array_shape_error)?;
389    let (array, memory) = array.into_parts();
390    let oid = crate::catalog::type_metadata::pg_type_oid(array_scalar_type(element)) as u32;
391    Ok(control.finish(Value::Array(array.with_element_type_oid(Some(oid))), memory)?)
392}
393
394fn array_elements(
395    values: &[Value],
396    element: &ColumnType,
397    control: &ProductionControl<'_>,
398) -> Result<Produced<Vec<Value>>> {
399    let mut output = ProductionVec::new(*control);
400    output.reserve(values.len())?;
401    for value in values {
402        let value = match value {
403            Value::List(values) => {
404                let (values, memory) = array_elements(values, element, control)?.into_parts();
405                control.finish(Value::List(values), memory)?
406            }
407            scalar => convert_value_to_column_type_with_control(
408                control.copy_value(scalar)?,
409                element,
410                control,
411            )?,
412        };
413        output.push_produced(value)?;
414    }
415    Ok(output.finish()?)
416}
417
418fn array_shape_error() -> SQLError {
419    SQLError::TypeMismatch("multidimensional arrays must have matching dimensions".into())
420}
421
422fn record_value(
423    value: Produced<Value>,
424    control: &ProductionControl<'_>,
425) -> Result<Produced<Value>> {
426    let Value::Row(source) = &*value else {
427        unreachable!();
428    };
429    // A record identity conversion must retain the producing expression's descriptor.
430    if source.field_types().is_some() {
431        return Ok(value);
432    }
433    let mut records = ProductionVec::new(*control);
434    records.reserve(source.len())?;
435    for index in 0..source.len() {
436        let (name, memory) = control.format(format_args!("f{}", index + 1))?.into_parts();
437        records.push_produced(control.finish((name, Value::Null), memory)?)?;
438    }
439    let (records, records_memory) = records.finish()?.into_parts();
440    let (Value::Row(source), source_memory) = value.into_parts() else {
441        unreachable!();
442    };
443    let old_buffer_bytes =
444        source.capacity() * size_of::<Value>() + uqa_core::RowValue::retained_header_bytes();
445    let mut parts = RecordParts {
446        source: source.into_values(),
447        records,
448        memory: control.combine(source_memory, records_memory),
449    };
450    for ((_, destination), source) in parts.records.iter_mut().zip(parts.source) {
451        control.check()?;
452        *destination = source;
453    }
454    if let Some(memory) = &mut parts.memory {
455        drop(memory.split(old_buffer_bytes));
456    }
457    let (record, memory) = uqa_core::RecordValue::with_control(
458        control.finish(parts.records, parts.memory)?,
459        None,
460        control,
461    )?
462    .into_parts();
463    Ok(control.finish(Value::Record(record), memory)?)
464}
465
466struct RecordParts {
467    source: Vec<Value>,
468    records: Vec<(String, Value)>,
469    memory: Option<MemoryReservation>,
470}
471
472fn vector_value(vector: &[f32], control: &ProductionControl<'_>) -> Result<Produced<Value>> {
473    let mut output = ProductionVec::new(*control);
474    output.reserve(vector.len())?;
475    for value in vector {
476        output.push_produced(
477            control.finish(Value::Float(f64::from(*value)), control.empty_reservation())?,
478        )?;
479    }
480    let (values, memory) = output.finish()?.into_parts();
481    Ok(control.finish(Value::List(values), memory)?)
482}
483
484fn parse_boolean_text(text: &str) -> Option<bool> {
485    let text = text.trim();
486    if ["true", "t", "yes", "y", "on", "1"]
487        .iter()
488        .any(|candidate| text.eq_ignore_ascii_case(candidate))
489    {
490        Some(true)
491    } else if ["false", "f", "no", "n", "off", "0"]
492        .iter()
493        .any(|candidate| text.eq_ignore_ascii_case(candidate))
494    {
495        Some(false)
496    } else {
497        None
498    }
499}
500
501#[cfg(test)]
502mod tests;
503
504/// `apply_typmod`: a value that does not fit the declared precision reports `22003` with the magnitude the field admits, `10^(precision - scale)`, or `1` when the precision equals the scale.
505#[must_use]
506pub fn numeric_field_overflow(precision: u32, scale: i32) -> SQLError {
507    let maxdigits = i64::from(precision) - i64::from(scale);
508    SQLError::Diagnostic {
509        sqlstate: "22003".into(),
510        message: "numeric field overflow".into(),
511        detail: Some(format!(
512            "A field with precision {precision}, scale {scale} must round to an absolute value less than {}.",
513            if maxdigits == 0 {
514                "1".to_string()
515            } else {
516                format!("10^{maxdigits}")
517            }
518        )),
519        hint: None,
520    }
521}
uqa_sql/assignment/conversion/production.rs

uqa_sql/assignment/conversion/
production.rs
