faucet_cli/
lineage_glue.rs

1//! Glue between the CLI config and `faucet-lineage`: builds the shared emitter
2//! and maps resolved transform specs onto column-lineage ops.
3
4use crate::config::TransformSpec;
5use faucet_core::FaucetError;
6use faucet_lineage::{ColumnOp, LineageConfig, LineageEmitter};
7use std::sync::Arc;
8
9/// Build the shared emitter from the parsed config. Returns `Ok(None)` when no
10/// `lineage:` block is present.
11pub fn build_emitter(
12    cfg: Option<&LineageConfig>,
13) -> Result<Option<Arc<LineageEmitter>>, FaucetError> {
14    match cfg {
15        Some(c) => Ok(Some(LineageEmitter::new(c.clone())?)),
16        None => Ok(None),
17    }
18}
19
20/// Best-effort transport reachability check (used by validate + doctor).
21/// Never fails a run — returns a human-readable Ok/Err string.
22pub async fn check_transport(cfg: &LineageConfig) -> Result<String, String> {
23    match &cfg.transport {
24        faucet_lineage::Transport::File { path } => {
25            let parent = path.parent().unwrap_or_else(|| std::path::Path::new("."));
26            if parent.exists() || tokio::fs::create_dir_all(parent).await.is_ok() {
27                Ok(format!("file path writable: {}", path.display()))
28            } else {
29                Err(format!("cannot create parent dir for {}", path.display()))
30            }
31        }
32        faucet_lineage::Transport::Http { url, .. } => {
33            match reqwest::Client::new().head(url).send().await {
34                Ok(_) => Ok(format!("http endpoint reachable: {url}")),
35                Err(e) => Err(format!("http endpoint unreachable: {e}")),
36            }
37        }
38        #[cfg(feature = "lineage-kafka")]
39        faucet_lineage::Transport::Kafka { brokers, .. } => {
40            Ok(format!("kafka brokers configured: {brokers} (not probed)"))
41        }
42    }
43}
44
45/// Map the resolved transform chain onto column-lineage ops. Transforms that
46/// change structure or rewrite keys (`flatten`, `explode`, `keys_case`,
47/// `rename_keys`) and any unknown transform become `Opaque`, which makes
48/// `faucet_lineage::derive` omit the column-lineage facet (never fabricated).
49///
50/// `has_masking` appends a trailing `Identity` op when a `masking:` policy is
51/// active: masking rewrites field *values* (redact/hash/tokenize/partial) but
52/// preserves the column set, so it is faithfully an identity at the
53/// column-lineage level — every output column derives from the same-named
54/// input column (#206).
55pub fn column_ops(specs: &[TransformSpec], has_masking: bool) -> Vec<ColumnOp> {
56    let mut ops: Vec<ColumnOp> = specs.iter().map(map_one).collect();
57    if has_masking {
58        ops.push(ColumnOp::Identity);
59    }
60    ops
61}
62
63fn map_one(s: &TransformSpec) -> ColumnOp {
64    match s.kind.as_str() {
65        "cast" | "redact" | "value_case" | "spell_symbols" => ColumnOp::Identity,
66        "select" => ColumnOp::Select(string_array(&s.config, "fields")),
67        "drop" => ColumnOp::Drop(string_array(&s.config, "fields")),
68        "set" => ColumnOp::Set(object_keys(&s.config, "values")),
69        "rename_field" => ColumnOp::Rename(string_pairs(&s.config, "fields")),
70        // structure-changing / key-rewriting / unknown
71        _ => ColumnOp::Opaque,
72    }
73}
74
75fn string_array(config: &serde_json::Value, key: &str) -> Vec<String> {
76    config
77        .get(key)
78        .and_then(|v| v.as_array())
79        .map(|a| {
80            a.iter()
81                .filter_map(|x| x.as_str().map(String::from))
82                .collect()
83        })
84        .unwrap_or_default()
85}
86
87fn object_keys(config: &serde_json::Value, key: &str) -> Vec<String> {
88    config
89        .get(key)
90        .and_then(|v| v.as_object())
91        .map(|m| m.keys().cloned().collect())
92        .unwrap_or_default()
93}
94
95fn string_pairs(config: &serde_json::Value, key: &str) -> Vec<(String, String)> {
96    config
97        .get(key)
98        .and_then(|v| v.as_object())
99        .map(|m| {
100            m.iter()
101                .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
102                .collect()
103        })
104        .unwrap_or_default()
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110    use crate::config::TransformSpec;
111    use serde_json::json;
112
113    fn spec(kind: &str, config: serde_json::Value) -> TransformSpec {
114        TransformSpec {
115            kind: kind.into(),
116            config,
117        }
118    }
119
120    #[test]
121    fn maps_explicit_mapping_transforms() {
122        let specs = vec![
123            spec("rename_field", json!({"fields": {"a": "b"}})),
124            spec("select", json!({"fields": ["b", "c"]})),
125            spec("cast", json!({"fields": {"b": "integer"}})),
126        ];
127        let ops = column_ops(&specs, false);
128        assert!(matches!(ops[0], faucet_lineage::ColumnOp::Rename(_)));
129        assert!(matches!(ops[1], faucet_lineage::ColumnOp::Select(_)));
130        assert!(matches!(ops[2], faucet_lineage::ColumnOp::Identity));
131    }
132
133    #[test]
134    fn masking_appends_trailing_identity_op() {
135        let specs = vec![spec("select", json!({"fields": ["a"]}))];
136        let ops = column_ops(&specs, true);
137        assert_eq!(ops.len(), 2);
138        assert!(matches!(ops[0], faucet_lineage::ColumnOp::Select(_)));
139        assert!(
140            matches!(ops[1], faucet_lineage::ColumnOp::Identity),
141            "masking is value-only + key-preserving → Identity"
142        );
143        // Without masking, no trailing op is appended.
144        assert_eq!(column_ops(&specs, false).len(), 1);
145    }
146
147    #[test]
148    fn maps_structure_changing_to_opaque() {
149        for k in [
150            "flatten",
151            "explode",
152            "keys_case",
153            "rename_keys",
154            "weird_custom",
155        ] {
156            let ops = column_ops(&[spec(k, json!({}))], false);
157            assert!(matches!(ops[0], faucet_lineage::ColumnOp::Opaque), "{k}");
158        }
159    }
160}
faucet_cli/lineage_glue.rs

faucet_cli/
lineage_glue.rs
