uqa-storage 0.5.2

Document store, inverted index, IVF/HNSW vectors, B-tree, R*Tree, catalog
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Controlled document decoding preserves each durable format's value semantics and transfers its field lease to immutable readers.

use uqa_core::{
    json::{JsonReader, JsonToken},
    memory::Budgeted,
};

use super::{
    migrate_legacy_stored_document, Document, DocumentMetadata, StorageBackendResult,
    DOCUMENT_VALUE_V1_PREFIX, DOCUMENT_VALUE_V2_PREFIX,
};
use crate::{
    document_store::{
        decoding::{
            decode_legacy_document_fields_budgeted, decoder, invalid_json, read_error,
            spans::ContainerSpans, text,
        },
        RetainedDocumentFields, RetainedStoredDocument,
    },
    read_control::StorageReadControl,
};

pub(in crate::key_value) fn decode_retained_stored_document_value(
    bytes: &[u8],
    control: &StorageReadControl,
) -> StorageBackendResult<RetainedStoredDocument> {
    control.check()?;
    let (fields, metadata) = if let Some(body) = bytes.strip_prefix(DOCUMENT_VALUE_V2_PREFIX) {
        current_fields(body, control)?
    } else {
        let fields = if let Some(body) = bytes.strip_prefix(DOCUMENT_VALUE_V1_PREFIX) {
            decoder(control).fields(text(body)?).map_err(read_error)?
        } else {
            decode_legacy_document_fields_budgeted(bytes, control)?
        };
        migrate(fields)?
    };
    let fields = RetainedDocumentFields::from_budgeted(fields, control)?;
    control.check()?;
    Ok(RetainedStoredDocument::with_metadata(fields, metadata))
}

fn current_fields(
    input: &[u8],
    control: &StorageReadControl,
) -> StorageBackendResult<(Budgeted<Document>, DocumentMetadata)> {
    let spans = ContainerSpans::read_envelope(input, control)?;
    let (fields, xmin) = match &spans {
        ContainerSpans::Object(members) => {
            let mut fields = None;
            let mut xmin = None;
            for member in members.iter() {
                control.check()?;
                match member.name.as_str() {
                    "fields" if fields.replace(member.value).is_some() => {
                        return Err(invalid_json("duplicate document fields"));
                    }
                    "tuple_xmin" if xmin.replace(member.value).is_some() => {
                        return Err(invalid_json("duplicate document tuple_xmin"));
                    }
                    // Unknown envelope fields are structurally validated by the shared reader without choosing a numeric or tagged-value representation.
                    _ => {}
                }
            }
            (
                fields.ok_or_else(|| invalid_json("missing document fields"))?,
                xmin,
            )
        }
        ContainerSpans::Array(values) if values.len() == 2 => (values[0], Some(values[1])),
        ContainerSpans::Array(_) => {
            return Err(invalid_json("document record needs two sequence elements"));
        }
    };
    let tuple_xmin = xmin
        .map(|value| decode_xmin(value, control))
        .transpose()?
        .flatten();
    let fields = decoder(control)
        .with_depth_limit(126)
        .fields(text(fields)?)
        .map_err(read_error)?;
    let metadata =
        tuple_xmin.map_or_else(DocumentMetadata::default, DocumentMetadata::with_tuple_xmin);
    Ok((fields, metadata))
}

fn decode_xmin(input: &[u8], control: &StorageReadControl) -> StorageBackendResult<Option<u32>> {
    let mut reader = JsonReader::from_slice(input, control.memory(), control.cancellation());
    let event = reader
        .next_event()
        .map_err(read_error)?
        .ok_or_else(|| invalid_json("missing tuple_xmin value"))?;
    let value = match event.token {
        JsonToken::Null => None,
        JsonToken::Number(number) => Some(
            number
                .parse::<u32>()
                .map_err(|_| invalid_json("tuple_xmin is outside the u32 range"))?,
        ),
        _ => return Err(invalid_json("tuple_xmin is not an integer or null")),
    };
    if reader.next_event().map_err(read_error)?.is_some() {
        return Err(invalid_json("trailing tuple_xmin value"));
    }
    Ok(value)
}

fn migrate(
    fields: Budgeted<Document>,
) -> StorageBackendResult<(Budgeted<Document>, DocumentMetadata)> {
    let (fields, memory) = fields.into_parts();
    let (fields, metadata) = migrate_legacy_stored_document(fields, true)?.into_parts();
    Ok((Budgeted::new(fields, memory), metadata))
}

#[cfg(test)]
mod tests;