uqa_sql/expr/
datums.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Read retained physical scalar bytes only when a SQL operation observes the value. Type inspection, copying and NULL tests do not invoke this decoder.
8
9use crate::SQLError;
10use uqa_core::{
11    memory::{Produced, ProductionControl, ProductionVec},
12    DatumValue, DecimalValue, Value,
13};
14
15mod arrays;
16mod compression;
17mod fixed;
18mod malformed;
19mod records;
20
21/// Whether catalog-free constant evaluation could reach an admitted physical value.
22pub fn contains_datum(value: &Value) -> bool {
23    match value {
24        Value::Datum(_) => true,
25        Value::Array(array) => array.elements().iter().any(contains_datum),
26        Value::List(values) => values.iter().any(contains_datum),
27        Value::Row(values) => values.iter().any(contains_datum),
28        Value::Record(fields) => fields.iter().any(|(_, value)| contains_datum(value)),
29        Value::Map(values) => values.values().any(contains_datum),
30        _ => false,
31    }
32}
33
34/// Read only the actual tuple type of a value already selected as a composite by SQL analysis.
35pub(in crate::expr) fn record_type_oid(
36    datum: &DatumValue,
37    control: &ProductionControl<'_>,
38) -> Result<u32, SQLError> {
39    let payload = payload(datum, control)?;
40    word(payload.bytes().get(4..).unwrap_or_default())
41        .ok_or_else(|| corrupt("invalid record header"))
42}
43
44/// Shape functions inspect only the detoasted array header, without resolving or observing its elements.
45pub(in crate::expr) fn array_shape(
46    datum: &DatumValue,
47    control: &ProductionControl<'_>,
48) -> Result<arrays::ArrayShape, SQLError> {
49    control.check()?;
50    let payload = payload(datum, control)?;
51    arrays::ArrayShape::read(payload.bytes())
52}
53
54#[cfg(test)]
55mod tests;
56
57pub fn type_name(datum: &DatumValue) -> &'static str {
58    match datum.type_oid() {
59        18 => "\"char\"",
60        19 => "name",
61        700 => "real",
62        1042 => "character",
63        1043 => "character varying",
64        oid => crate::catalog::type_metadata::catalog_type_name(i64::from(oid)),
65    }
66}
67
68pub fn read(datum: &DatumValue) -> Result<Value, SQLError> {
69    read_with_catalog(datum, None)
70}
71
72/// Read an admitted physical value through the current statement's catalog, without invoking type input or domain checks.
73pub fn read_with_catalog(
74    datum: &DatumValue,
75    engine: Option<&dyn super::EngineHook>,
76) -> Result<Value, SQLError> {
77    read_with_catalog_and_control(datum, engine, &ProductionControl::uncontrolled())
78        .map(|value| value.into_uncontrolled().expect("ordinary datum read"))
79}
80
81/// Copy a selected constant field at planning time, detoasting variable-width storage without interpreting its scalar contents. Whole records remain deferred.
82pub fn copy_constant_field(value: &Value, ty: &crate::ColumnType) -> Result<Value, SQLError> {
83    let Value::Datum(datum) = value else {
84        return Ok(value.clone());
85    };
86    if crate::catalog::type_metadata::pg_type_len(ty) != -1 {
87        return Ok(value.clone());
88    }
89    let control = ProductionControl::uncontrolled();
90    let payload = payload(datum, &control)?;
91    let size = payload
92        .bytes()
93        .len()
94        .checked_add(4)
95        .and_then(|size| u32::try_from(size).ok())
96        .and_then(|size| size.checked_mul(4))
97        .ok_or_else(|| corrupt("invalid datum length"))?;
98    let mut bytes = size.to_le_bytes().to_vec();
99    bytes.extend_from_slice(payload.bytes());
100    Ok(Value::Datum(DatumValue::new(datum.type_oid(), 0, bytes)))
101}
102
103pub fn read_with_control(
104    datum: &DatumValue,
105    control: &ProductionControl<'_>,
106) -> Result<Produced<Value>, SQLError> {
107    read_with_catalog_and_control(datum, None, control)
108}
109
110/// Preserve the caller's output allowance and cancellation boundary during catalog-dependent physical reads.
111pub fn read_with_catalog_and_control(
112    datum: &DatumValue,
113    engine: Option<&dyn super::EngineHook>,
114    control: &ProductionControl<'_>,
115) -> Result<Produced<Value>, SQLError> {
116    let catalog = engine.map(super::value_catalog::EngineValueCatalog);
117    read_with_value_catalog_and_control(
118        datum,
119        catalog
120            .as_ref()
121            .map(|catalog| catalog as &dyn super::SQLValueCatalog),
122        control,
123    )
124}
125
126/// Decode a reached value through the same borrowed catalog used by SQL comparison, hashing and output.
127pub fn read_with_value_catalog_and_control(
128    datum: &DatumValue,
129    catalog: Option<&dyn super::SQLValueCatalog>,
130    control: &ProductionControl<'_>,
131) -> Result<Produced<Value>, SQLError> {
132    control.check()?;
133    if let Some(value) = fixed::read(datum, control) {
134        return value;
135    }
136    if crate::catalog::type_metadata::builtin_scalar_type(datum.type_oid()).is_none()
137        && crate::catalog::type_metadata::builtin_array_element(datum.type_oid()).is_none()
138    {
139        if let Some(enums) = catalog {
140            if enums.enum_type_labels(datum.type_oid())?.is_some() {
141                let oid = enum_label_oid(datum)?;
142                return Ok(
143                    control.retain_external_value(super::enums::enum_value_from_oid(
144                        Some(enums),
145                        oid,
146                    )?)?,
147                );
148            }
149        }
150    }
151    let payload = payload(datum, control)?;
152    read_payload(datum.type_oid(), payload.bytes(), catalog, control)
153}
154
155/// Read the fixed-width OID after the caller has selected an enum operation.
156/// Label existence belongs to the particular consuming operation, not this read.
157pub(crate) fn enum_label_oid(datum: &DatumValue) -> Result<u32, SQLError> {
158    word(
159        datum
160            .bytes()
161            .get(datum.offset() as usize..)
162            .unwrap_or_default(),
163    )
164    .ok_or_else(|| corrupt("invalid datum length"))
165}
166
167pub(super) fn compare_jsonb_with_control(
168    left: &Value,
169    right: &Value,
170    control: &ProductionControl<'_>,
171) -> Result<Option<std::cmp::Ordering>, SQLError> {
172    let physical = |value: &Value| matches!(value, Value::Datum(datum) if datum.type_oid() == 3802);
173    let jsonb = |value: &Value| physical(value) || matches!(value, Value::JsonB(_));
174    if !(jsonb(left) && jsonb(right) && (physical(left) || physical(right))) {
175        return Ok(None);
176    }
177    let left_bytes = match left {
178        Value::Datum(datum) => Some(payload(datum, control)?),
179        _ => None,
180    };
181    let right_bytes = match right {
182        Value::Datum(datum) => Some(payload(datum, control)?),
183        _ => None,
184    };
185    let left = match left {
186        Value::JsonB(text) => super::json::JsonbInput::Text(text),
187        _ => super::json::JsonbInput::Bytes(left_bytes.as_ref().expect("physical JSONB").bytes()),
188    };
189    let right = match right {
190        Value::JsonB(text) => super::json::JsonbInput::Text(text),
191        _ => super::json::JsonbInput::Bytes(right_bytes.as_ref().expect("physical JSONB").bytes()),
192    };
193    super::json::compare_jsonb_datums_with_control(left, right, control).map(Some)
194}
195
196fn read_payload(
197    oid: u32,
198    bytes: &[u8],
199    catalog: Option<&dyn super::SQLValueCatalog>,
200    control: &ProductionControl<'_>,
201) -> Result<Produced<Value>, SQLError> {
202    if crate::catalog::type_metadata::builtin_array_element(oid).is_some() {
203        return arrays::read(bytes, catalog, control);
204    }
205    if crate::catalog::type_metadata::builtin_scalar_type(oid).is_none() {
206        if let Some(ty) = catalog
207            .map(|catalog| catalog.value_type_by_oid(oid))
208            .transpose()?
209            .flatten()
210        {
211            match ty {
212                crate::ColumnType::Array(_) => return arrays::read(bytes, catalog, control),
213                crate::ColumnType::Composite(_) => return records::read(bytes, catalog, control),
214                crate::ColumnType::Domain { base, .. } => {
215                    return read_payload(
216                        crate::catalog::type_metadata::pg_type_oid(&base) as u32,
217                        bytes,
218                        catalog,
219                        control,
220                    )
221                }
222                _ => {}
223            }
224        }
225    }
226    match oid {
227        17 => {
228            let mut output = ProductionVec::new(*control);
229            output.reserve(bytes.len())?;
230            for byte in bytes {
231                output.push_copy(*byte)?;
232            }
233            let (bytes, memory) = output.finish()?.into_parts();
234            Ok(control.finish(Value::Bytes(bytes), memory)?)
235        }
236        25 | 114 | 194 | 1042 | 1043 | 1790 => {
237            let text = std::str::from_utf8(bytes).map_err(|error| SQLError::Routine {
238                sqlstate: "22021".into(),
239                message: format!(
240                    "invalid byte sequence for encoding \"UTF8\": 0x{:02x}",
241                    bytes[error.valid_up_to()]
242                ),
243            })?;
244            let (text, memory) = control.copy_text(text)?.into_parts();
245            let value = match oid {
246                114 => Value::Json(text),
247                1042 => Value::FixedChar(text),
248                _ => Value::Str(text),
249            };
250            Ok(control.finish(value, memory)?)
251        }
252        3802 => {
253            let (text, memory) =
254                super::json::decode_jsonb_datum_with_control(bytes, control)?.into_parts();
255            Ok(control.finish(Value::JsonB(text), memory)?)
256        }
257        1700 => {
258            let text =
259                crate::catalog::node_tree::decode_numeric_datum_with_control(bytes, control)?;
260            let (number, memory) = DecimalValue::parse_with_control(&text, control)?
261                .ok_or_else(|| corrupt("invalid numeric datum"))?
262                .into_parts();
263            Ok(control.finish(Value::Decimal(number), memory)?)
264        }
265        oid => Err(SQLError::Unsupported(format!(
266            "physical datum output for type OID {oid}"
267        ))),
268    }
269}
270
271/// Borrow ordinary and inline physical bytea payloads; compressed input retains its admitted decompression buffer through consumption.
272pub(super) fn binary_payload<'a>(
273    value: &'a Value,
274    control: &ProductionControl<'_>,
275) -> Result<Option<Payload<'a>>, SQLError> {
276    control.check()?;
277    match value {
278        Value::Bytes(bytes) => Ok(Some(Payload::Borrowed(bytes))),
279        Value::Datum(datum) if datum.type_oid() == 17 => payload(datum, control).map(Some),
280        _ => Ok(None),
281    }
282}
283
284/// Binary length observes the raw-size header and does not decompress the payload.
285pub(super) fn binary_length(
286    value: &Value,
287    control: &ProductionControl<'_>,
288) -> Result<Option<i64>, SQLError> {
289    control.check()?;
290    let datum = match value {
291        Value::Bytes(bytes) => return Ok(Some(bytes.len() as i64)),
292        Value::Datum(datum) if datum.type_oid() == 17 => datum,
293        _ => return Ok(None),
294    };
295    let bytes = datum
296        .bytes()
297        .get(datum.offset() as usize..)
298        .ok_or_else(|| corrupt("invalid datum length"))?;
299    let first = *bytes
300        .first()
301        .ok_or_else(|| corrupt("invalid datum length"))?;
302    let size = if first == 1 && bytes.get(1) == Some(&18) {
303        i64::from(
304            word(bytes.get(2..).unwrap_or_default())
305                .ok_or_else(|| corrupt("invalid datum length"))?,
306        ) - 4
307    } else if first & 1 != 0 {
308        i64::from(first >> 1) - 1
309    } else if first & 3 == 2 {
310        i64::from(
311            word(bytes.get(4..).unwrap_or_default())
312                .ok_or_else(|| corrupt("invalid datum length"))?
313                & 0x3fff_ffff,
314        )
315    } else {
316        i64::from(word(bytes).ok_or_else(|| corrupt("invalid datum length"))? >> 2) - 4
317    };
318    Ok(Some(size))
319}
320
321pub(super) enum Payload<'a> {
322    Borrowed(&'a [u8]),
323    Owned(Produced<Vec<u8>>),
324}
325
326impl Payload<'_> {
327    pub(super) fn bytes(&self) -> &[u8] {
328        match self {
329            Self::Borrowed(bytes) => bytes,
330            Self::Owned(bytes) => bytes,
331        }
332    }
333}
334
335fn payload<'a>(
336    datum: &'a DatumValue,
337    control: &ProductionControl<'_>,
338) -> Result<Payload<'a>, SQLError> {
339    let offset = datum.offset() as usize;
340    let bytes = datum
341        .bytes()
342        .get(offset..)
343        .ok_or_else(|| corrupt("invalid datum length"))?;
344    let first = *bytes
345        .first()
346        .ok_or_else(|| corrupt("invalid datum length"))?;
347    if first == 1 {
348        return Err(malformed::external(datum));
349    }
350    if first & 1 != 0 {
351        let length = usize::from(first >> 1);
352        return bytes
353            .get(1..length)
354            .map(Payload::Borrowed)
355            .ok_or_else(|| corrupt("invalid datum length"));
356    }
357    let header = word(bytes).ok_or_else(|| corrupt("invalid datum length"))?;
358    let length = (header >> 2) as usize;
359    if header & 3 == 2 {
360        let information = word(bytes.get(4..).unwrap_or_default()).unwrap_or(0);
361        let method = information >> 30;
362        let compressed = bytes
363            .get(8..length)
364            .ok_or_else(|| compression::corrupt(method))?;
365        return compression::decompress(
366            compressed,
367            (information & 0x3fff_ffff) as usize,
368            method,
369            control,
370        )
371        .map(Payload::Owned);
372    }
373    bytes
374        .get(4..length)
375        .map(Payload::Borrowed)
376        .ok_or_else(|| corrupt("invalid datum length"))
377}
378
379fn word(bytes: &[u8]) -> Option<u32> {
380    Some(u32::from_le_bytes(bytes.get(..4)?.try_into().ok()?))
381}
382
383fn corrupt(message: &str) -> SQLError {
384    SQLError::Routine {
385        sqlstate: "XX001".into(),
386        message: message.into(),
387    }
388}
uqa_sql/expr/datums.rs

uqa_sql/expr/
datums.rs
