laminar_connectors/fpml/profile/
mod.rs

1//! Projection profiles: which elements become rows and how values map to typed Arrow columns.
2//!
3//! A profile is a TOML file with a `[profile]` header, `[[columns]]`, and optional
4//! `[namespaces]`, `[sets]`, and `[structs.<template>]` sections. `include` pulls in other files
5//! (relative to the including file, or `bundled:<path>` for a bundled file) that contribute
6//! namespaces, sets, struct templates, and columns; an included file's columns come before the
7//! including file's own, in include order, and a file included twice contributes once. Profiles
8//! are compiled against the pinned schema closure before activation: unknown element or attribute
9//! names, unknown templates, and incompatible types fail configuration, never ingestion.
10
11mod bundled;
12mod compile;
13mod plan;
14mod selector;
15mod spec;
16#[cfg(test)]
17mod tests;
18mod types;
19
20use std::collections::BTreeSet;
21use std::path::{Path, PathBuf};
22use std::sync::Arc;
23
24use sha2::{Digest, Sha256};
25
26pub(crate) use plan::{
27    Alternative, Column, Conversion, Meta, NameTest, Plan, Selector, Source, Step, Terminal,
28};
29pub(crate) use types::UTC_ZONE;
30
31use self::compile::Merged;
32use self::spec::ProfileFile;
33use super::artifacts::{hex, Artifact};
34use super::schema_model::SchemaModel;
35use crate::error::ConnectorError;
36use crate::schema::resolution::NativeSchema;
37
38pub use bundled::BUNDLED_PROFILES;
39
40const BUNDLED_PREFIX: &str = "bundled:";
41const MAX_INCLUDE_DEPTH: usize = 8;
42const MAX_PROFILE_FILES: usize = 256;
43const MAX_PROFILE_FILE_BYTES: u64 = 4 * 1024 * 1024;
44
45/// Where a profile comes from.
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub enum ProfileSource {
48    /// A profile shipped with LaminarDB, by name (see [`BUNDLED_PROFILES`]).
49    Bundled(String),
50    /// An operator-supplied profile file.
51    File(PathBuf),
52}
53
54/// A compiled profile with its identity.
55#[derive(Debug, Clone)]
56pub struct Profile {
57    /// Profile name from its header.
58    pub name: String,
59    /// Profile version from its header.
60    pub version: u32,
61    /// Header description.
62    pub description: String,
63    /// SHA-256 over the closure fingerprint and every profile file, in load order.
64    pub fingerprint: String,
65    pub(crate) plan: Arc<Plan>,
66}
67
68impl Profile {
69    /// Load, merge, and compile `source` against `artifact`.
70    ///
71    /// # Errors
72    /// Returns a configuration error for an unreadable, malformed, or incompatible profile.
73    pub fn load(source: &ProfileSource, artifact: &Artifact) -> Result<Self, ConnectorError> {
74        let mut loader = Loader::default();
75        let root = match source {
76            ProfileSource::Bundled(name) => {
77                let path = bundled::root(name).ok_or_else(|| {
78                    config_error(format!(
79                        "unknown bundled FpML profile '{name}' (available: {})",
80                        BUNDLED_PROFILES
81                            .iter()
82                            .map(|(name, _)| *name)
83                            .collect::<Vec<_>>()
84                            .join(", ")
85                    ))
86                })?;
87                Location::Bundled(path.to_owned())
88            }
89            ProfileSource::File(path) => Location::Disk(path.clone()),
90        };
91        let mut merged = Merged::default();
92        loader.load(&root, "<root>".into(), 0, true, &mut merged)?;
93        let model = SchemaModel::for_artifact(artifact)?;
94        let plan = compile::compile(&merged, artifact, &model).map_err(config_error)?;
95        let header = merged
96            .header
97            .as_ref()
98            .ok_or_else(|| config_error("the root profile file has no [profile] header".into()))?;
99        let mut digest = Sha256::new();
100        digest.update(artifact.fingerprint.as_bytes());
101        for (name, content) in &loader.files {
102            digest.update(name.as_bytes());
103            digest.update([0]);
104            digest.update(content.as_bytes());
105            digest.update([0]);
106        }
107        Ok(Self {
108            name: header.name.clone(),
109            version: header.version,
110            description: header.description.clone(),
111            fingerprint: hex(&digest.finalize()),
112            plan: Arc::new(plan),
113        })
114    }
115
116    /// The native schema contract recorded in a source's schema binding: the closure and
117    /// profile identities a restore must find unchanged.
118    #[must_use]
119    pub fn native_schema(&self, artifact: &Artifact) -> NativeSchema {
120        let mut identity = std::collections::BTreeMap::from([
121            ("schema_closure".to_owned(), artifact.id.clone()),
122            (
123                "schema_closure_sha256".to_owned(),
124                artifact.fingerprint.clone(),
125            ),
126            ("fpml_release".to_owned(), artifact.release.clone()),
127            ("fpml_view".to_owned(), artifact.view.clone()),
128            ("profile".to_owned(), self.name.clone()),
129            ("profile_version".to_owned(), self.version.to_string()),
130            ("profile_sha256".to_owned(), self.fingerprint.clone()),
131        ]);
132        if let Some(build) = artifact.build {
133            identity.insert("fpml_build".to_owned(), build.to_string());
134        }
135        NativeSchema {
136            format: "fpml".to_owned(),
137            identity,
138            definition: serde_json::json!({"description": self.description}),
139            references: Vec::new(),
140        }
141    }
142
143    /// Whether every document yields exactly one row: the rows selector is the document root.
144    #[must_use]
145    pub fn one_row_per_document(&self) -> bool {
146        matches!(self.plan.rows.alternatives.as_slice(), [only] if only.steps.is_empty())
147    }
148
149    /// The Arrow schema every batch from this profile has.
150    #[must_use]
151    pub fn schema(&self) -> arrow_schema::SchemaRef {
152        Arc::clone(&self.plan.schema)
153    }
154}
155
156#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
157enum Location {
158    Bundled(String),
159    Disk(PathBuf),
160}
161
162#[derive(Default)]
163struct Loader {
164    files: Vec<(String, String)>,
165    visited: BTreeSet<Location>,
166}
167
168impl Loader {
169    /// `identity` names the file in the fingerprint: the bundled path, or the include text as
170    /// written, so moving an operator profile directory does not change its fingerprint.
171    fn load(
172        &mut self,
173        location: &Location,
174        identity: String,
175        depth: usize,
176        root: bool,
177        merged: &mut Merged,
178    ) -> Result<(), ConnectorError> {
179        if depth > MAX_INCLUDE_DEPTH {
180            return Err(config_error(format!(
181                "profile includes nest deeper than {MAX_INCLUDE_DEPTH}"
182            )));
183        }
184        if !self.visited.insert(location.clone()) {
185            return Ok(());
186        }
187        if self.files.len() >= MAX_PROFILE_FILES {
188            return Err(config_error(format!(
189                "a profile may load at most {MAX_PROFILE_FILES} files"
190            )));
191        }
192        let (name, content) = read(location)?;
193        let file: ProfileFile = toml::from_str(&content)
194            .map_err(|error| config_error(format!("profile file '{name}': {error}")))?;
195        self.files.push((identity, content));
196        let located = |message: String| config_error(format!("profile file '{name}': {message}"));
197        match (root, file.profile) {
198            (true, Some(header)) => merged.header = Some(header),
199            (true, None) => return Err(located("the root file needs a [profile] header".into())),
200            (false, Some(_)) => {
201                return Err(located(
202                    "an included file must not have a [profile] header".into(),
203                ))
204            }
205            (false, None) => {}
206        }
207        for (prefix, uri) in file.namespaces {
208            if let Some(existing) = merged.namespaces.get(&prefix) {
209                if existing != &uri {
210                    return Err(located(format!(
211                        "prefix '{prefix}' is bound to '{existing}' and '{uri}'"
212                    )));
213                }
214            }
215            merged.namespaces.insert(prefix, uri);
216        }
217        for (set, spec) in file.sets {
218            if merged
219                .sets
220                .get(&set)
221                .is_some_and(|existing| existing != &spec)
222            {
223                return Err(located(format!("set '{set}' is defined twice differently")));
224            }
225            merged.sets.insert(set, spec);
226        }
227        for (template, spec) in file.structs {
228            if merged
229                .structs
230                .get(&template)
231                .is_some_and(|existing| existing != &spec)
232            {
233                return Err(located(format!(
234                    "struct template '{template}' is defined twice differently"
235                )));
236            }
237            merged.structs.insert(template, spec);
238        }
239        for include in &file.include {
240            let next = resolve(location, include).map_err(located)?;
241            let identity = match &next {
242                Location::Bundled(path) => format!("{BUNDLED_PREFIX}{path}"),
243                Location::Disk(_) => include.clone(),
244            };
245            self.load(&next, identity, depth + 1, false, merged)?;
246        }
247        merged.columns.extend(file.columns);
248        Ok(())
249    }
250}
251
252fn resolve(from: &Location, include: &str) -> Result<Location, String> {
253    if let Some(path) = include.strip_prefix(BUNDLED_PREFIX) {
254        return bundled::file(path)
255            .map(|_| Location::Bundled(path.to_owned()))
256            .ok_or_else(|| format!("unknown bundled profile file '{path}'"));
257    }
258    if Path::new(include).is_absolute() || include.contains('\\') {
259        return Err(format!(
260            "include '{include}' must be a relative '/'-separated path"
261        ));
262    }
263    match from {
264        Location::Bundled(path) => {
265            let joined = bundled::join(path, include)?;
266            bundled::file(&joined)
267                .map(|_| Location::Bundled(joined.clone()))
268                .ok_or_else(|| format!("unknown bundled profile file '{joined}'"))
269        }
270        Location::Disk(path) => Ok(Location::Disk(
271            path.parent()
272                .unwrap_or_else(|| Path::new("."))
273                .join(include),
274        )),
275    }
276}
277
278fn read(location: &Location) -> Result<(String, String), ConnectorError> {
279    match location {
280        Location::Bundled(path) => bundled::file(path)
281            .map(|content| (format!("{BUNDLED_PREFIX}{path}"), content.to_owned()))
282            .ok_or_else(|| config_error(format!("unknown bundled profile file '{path}'"))),
283        Location::Disk(path) => {
284            let size = std::fs::metadata(path)
285                .map_err(|error| {
286                    config_error(format!("cannot read profile '{}': {error}", path.display()))
287                })?
288                .len();
289            if size > MAX_PROFILE_FILE_BYTES {
290                return Err(config_error(format!(
291                    "profile '{}' exceeds {MAX_PROFILE_FILE_BYTES} bytes",
292                    path.display()
293                )));
294            }
295            let content = std::fs::read_to_string(path).map_err(|error| {
296                config_error(format!("cannot read profile '{}': {error}", path.display()))
297            })?;
298            Ok((path.display().to_string(), content))
299        }
300    }
301}
302
303fn config_error(message: String) -> ConnectorError {
304    ConnectorError::ConfigurationError(message)
305}
laminar_connectors/fpml/profile/mod.rs

laminar_connectors/fpml/profile/
mod.rs
