faucet_core/
discover.rs

1//! Live source introspection (`faucet discover`, #211).
2//!
3//! A [`Source`](crate::Source) that can enumerate the datasets living behind
4//! its connection (tables in a database schema, collections in MongoDB,
5//! indices in Elasticsearch, key prefixes in an object store) overrides
6//! [`Source::discover`](crate::Source::discover) to return one
7//! [`DatasetDescriptor`] per dataset. The CLI turns that list into a
8//! ready-to-run config: one matrix row per dataset, each row deep-merging the
9//! descriptor's [`config_patch`](DatasetDescriptor::config_patch) over the
10//! connection config.
11
12use serde::{Deserialize, Serialize};
13use serde_json::{Value, json};
14
15/// One dataset discovered behind a source's connection.
16#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
17pub struct DatasetDescriptor {
18    /// Human-readable dataset identity — a qualified table name
19    /// (`public.orders`), a collection, an index, or an object-key prefix.
20    pub name: String,
21    /// What kind of dataset this is: `table`, `collection`, `index`,
22    /// `prefix`, or `object`. Informational (rendered in output); the
23    /// selection mechanics live entirely in [`config_patch`](Self::config_patch).
24    pub kind: String,
25    /// The dataset's column shape as an
26    /// [`infer_schema`](crate::schema::infer_schema)-shaped object
27    /// (`{"type":"object","properties":{…}}`), or `None` when the source
28    /// cannot cheaply introspect it (object stores, schemaless collections
29    /// with no sample).
30    #[serde(default, skip_serializing_if = "Option::is_none")]
31    pub schema: Option<Value>,
32    /// Approximate row/object count when the catalog exposes one cheaply
33    /// (e.g. `pg_class.reltuples`). Never computed via a full scan.
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub estimated_rows: Option<u64>,
36    /// Partial source-config **override** selecting exactly this dataset —
37    /// deep-merged over the connection config by a matrix row (e.g.
38    /// `{"query": "SELECT * FROM \"public\".\"orders\""}` for a SQL source,
39    /// `{"collection": "orders"}` for MongoDB, `{"prefix": "raw/orders/"}`
40    /// for an object store). Must never contain credentials.
41    pub config_patch: Value,
42}
43
44impl DatasetDescriptor {
45    /// Construct a descriptor with no schema / row estimate.
46    pub fn new(name: impl Into<String>, kind: impl Into<String>, config_patch: Value) -> Self {
47        Self {
48            name: name.into(),
49            kind: kind.into(),
50            schema: None,
51            estimated_rows: None,
52            config_patch,
53        }
54    }
55
56    /// Attach an inferred/introspected schema.
57    pub fn with_schema(mut self, schema: Value) -> Self {
58        self.schema = Some(schema);
59        self
60    }
61
62    /// Attach a catalog-provided row estimate.
63    pub fn with_estimated_rows(mut self, rows: u64) -> Self {
64        self.estimated_rows = Some(rows);
65        self
66    }
67}
68
69/// Map a SQL catalog type name (as reported by `information_schema.columns`
70/// or an equivalent) to a JSON-Schema type fragment matching the shape
71/// [`infer_schema`](crate::schema::infer_schema) produces.
72///
73/// The mapping is deliberately conservative and dialect-tolerant — it matches
74/// on lowercase substrings so `TIMESTAMP WITH TIME ZONE`, `timestamptz`,
75/// `Nullable(Int64)` etc. all land on a sensible JSON type. Unknown types map
76/// to `string` (every SQL value has a textual form, so `string` is the safe
77/// over-approximation — never a lossy one like `number`).
78pub fn sql_type_to_json_schema(data_type: &str) -> Value {
79    let t = data_type.to_ascii_lowercase();
80    let ty = if t.contains("bool") || t == "bit" {
81        "boolean"
82    } else if t.contains("json") || t.contains("variant") || t == "object" || t.contains("struct") {
83        // checked before the integer branch so `STRUCT<a INT>` maps to object
84        "object"
85    } else if t.contains("array") || t.starts_with('_') {
86        // Postgres reports array types as `ARRAY` (information_schema) or
87        // `_int4`-style internal names.
88        "array"
89    } else if (t.contains("int") && !t.contains("interval") && !t.contains("point"))
90        || t == "serial"
91        || t == "bigserial"
92        || t == "smallserial"
93    {
94        // covers int2/4/8, integer, bigint, smallint, tinyint, mediumint —
95        // but not `interval` / `point`, whose substring would false-match
96        "integer"
97    } else if t.contains("double")
98        || t.contains("float")
99        || t.contains("real")
100        || t.contains("decimal")
101        || t.contains("numeric")
102        || t.contains("money")
103        || t == "number"
104    {
105        "number"
106    } else {
107        // text, varchar, char, uuid, date, time, timestamp, bytea, blob,
108        // enum, inet, … — all serialized as JSON strings by the sources.
109        "string"
110    };
111    json!({ "type": ty })
112}
113
114/// Wrap a type fragment as nullable (`{"type": ["T", "null"]}`), matching the
115/// nullable shape [`infer_schema`](crate::schema::infer_schema) emits.
116pub fn nullable_type(fragment: Value) -> Value {
117    match fragment.get("type") {
118        Some(Value::String(t)) => json!({ "type": [t, "null"] }),
119        _ => fragment,
120    }
121}
122
123/// Assemble an `infer_schema`-shaped object schema from
124/// `(column_name, type_fragment)` pairs, preserving input order semantics
125/// (`serde_json` map ordering applies).
126pub fn columns_to_schema(columns: impl IntoIterator<Item = (String, Value)>) -> Value {
127    let mut properties = serde_json::Map::new();
128    for (name, fragment) in columns {
129        properties.insert(name, fragment);
130    }
131    json!({ "type": "object", "properties": Value::Object(properties) })
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137
138    #[test]
139    fn descriptor_builder_round_trips() {
140        let d = DatasetDescriptor::new("public.orders", "table", json!({"query": "SELECT 1"}))
141            .with_schema(json!({"type": "object", "properties": {}}))
142            .with_estimated_rows(42);
143        assert_eq!(d.name, "public.orders");
144        assert_eq!(d.kind, "table");
145        assert_eq!(d.estimated_rows, Some(42));
146        let v = serde_json::to_value(&d).unwrap();
147        let back: DatasetDescriptor = serde_json::from_value(v).unwrap();
148        assert_eq!(back, d);
149    }
150
151    #[test]
152    fn descriptor_omits_absent_optionals_in_json() {
153        let d = DatasetDescriptor::new("t", "table", json!({}));
154        let v = serde_json::to_value(&d).unwrap();
155        assert!(v.get("schema").is_none());
156        assert!(v.get("estimated_rows").is_none());
157    }
158
159    #[test]
160    fn sql_types_map_to_json_types() {
161        for (sql, want) in [
162            ("integer", "integer"),
163            ("BIGINT", "integer"),
164            ("smallint", "integer"),
165            ("tinyint(1)", "integer"),
166            ("serial", "integer"),
167            ("double precision", "number"),
168            ("NUMERIC(10,2)", "number"),
169            ("decimal", "number"),
170            ("float8", "number"),
171            ("money", "number"),
172            ("boolean", "boolean"),
173            ("BOOL", "boolean"),
174            ("bit", "boolean"),
175            ("json", "object"),
176            ("JSONB", "object"),
177            ("VARIANT", "object"),
178            ("STRUCT<a INT>", "object"),
179            ("ARRAY", "array"),
180            ("_int4", "array"),
181            ("text", "string"),
182            ("character varying", "string"),
183            ("uuid", "string"),
184            ("timestamp with time zone", "string"),
185            ("date", "string"),
186            ("bytea", "string"),
187            ("some_exotic_type", "string"),
188        ] {
189            assert_eq!(
190                sql_type_to_json_schema(sql),
191                json!({ "type": want }),
192                "for SQL type {sql:?}"
193            );
194        }
195    }
196
197    #[test]
198    fn nullable_wraps_scalar_type() {
199        assert_eq!(
200            nullable_type(json!({"type": "integer"})),
201            json!({"type": ["integer", "null"]})
202        );
203        // Already-complex fragments pass through untouched.
204        let complex = json!({"type": ["string", "null"]});
205        assert_eq!(nullable_type(complex.clone()), complex);
206    }
207
208    #[test]
209    fn columns_to_schema_shapes_like_infer_schema() {
210        let schema = columns_to_schema(vec![
211            ("id".to_string(), json!({"type": "integer"})),
212            ("name".to_string(), json!({"type": ["string", "null"]})),
213        ]);
214        assert_eq!(schema["type"], "object");
215        assert_eq!(schema["properties"]["id"]["type"], "integer");
216        assert_eq!(schema["properties"]["name"]["type"][1], "null");
217    }
218}
faucet_core/discover.rs

faucet_core/
discover.rs
