facett-lineage/src/openlineage.rs
//! **OpenLineage / OpenMetadata ingest** — render lineage that arrived over the
//! open *standard* wire (an OpenLineage `RunEvent` stream), not only skade's
//! reserved `lineage_events` Arrow table.
//!
//! This is the exact inverse of skade's `HttpLineageSink::run_event` mapping
//! (skade → OpenLineage), and it is the format OpenMetadata / Marquez / DataHub
//! all speak: a skade commit POSTed to an OpenLineage collector (or an
//! OpenMetadata OpenLineage ingestion endpoint) can be read straight back into
//! the SAME [`LineageRow`]s the Arrow path produces, so one viewer renders both
//! the skade-direct feed and the standard-collector feed identically.
//!
//! A [`RunEvent`](https://openlineage.io/spec) carries: `run.runId`,
//! `job.{namespace,name}`, `eventTime`, `inputs[]`/`outputs[]` (each a dataset
//! `{namespace,name,facets}`). We flatten it into one [`LineageRow`] per declared
//! input (matching the reserved-table fan-out), or one input-less row when the
//! event declares none. Conventions mirror skade's emitter:
//!
//! * dataset `namespace == job.namespace` ⇒ an in-warehouse dataset (`system` =
//! `None`); any other namespace ⇒ a foreign system whose *instance* is that
//! namespace (kind defaults to `"external"`, since OpenLineage folds the system
//! kind into the namespace string).
//! * `outputs[0].facets.version.datasetVersion` / `inputs[i].facets.version…` ⇒
//! the snapshot id.
//! * `run.facets.skade_commit.commitSeq` ⇒ the catalog commit cursor.
//! * the operation rides skade's `job.facets.documentation.description`
//! (`"skade <op> on <table>"`); absent that, it defaults to `"append"`.
use crate::LineageRow;
/// Parse a single OpenLineage **`RunEvent`** JSON value into [`LineageRow`]s (one
/// per declared input, or one input-less row). Unknown/optional fields degrade
/// gracefully; only a usable `outputs[0].name` is required.
///
/// # Errors
/// Returns a message if the value is not an object or carries no output dataset
/// with a `name`.
pub fn read_run_event(ev: &serde_json::Value) -> Result<Vec<LineageRow>, String> {
let obj = ev.as_object().ok_or("OpenLineage event is not a JSON object")?;
let job_ns = obj
.get("job")
.and_then(|j| j.get("namespace"))
.and_then(|v| v.as_str())
.unwrap_or("");
let actor = obj
.get("job")
.and_then(|j| j.get("name"))
.and_then(|v| v.as_str())
.unwrap_or("skade")
.to_string();
let event_id = obj
.get("run")
.and_then(|r| r.get("runId"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let ts_micros = obj
.get("eventTime")
.and_then(|v| v.as_str())
.and_then(rfc3339_to_micros)
.unwrap_or(0);
let commit_seq = obj
.get("run")
.and_then(|r| r.get("facets"))
.and_then(|f| f.get("skade_commit"))
.and_then(|c| c.get("commitSeq"))
.and_then(serde_json::Value::as_i64);
let operation = obj
.get("job")
.and_then(|j| j.get("facets"))
.and_then(|f| f.get("documentation"))
.and_then(|d| d.get("description"))
.and_then(|v| v.as_str())
.and_then(parse_operation)
.unwrap_or_else(|| "append".to_string());
// The produced dataset (skade emits exactly one output per event).
let out = obj
.get("outputs")
.and_then(|v| v.as_array())
.and_then(|a| a.first())
.ok_or("OpenLineage event has no output dataset")?;
let (output_table, target_system, target_instance, output_snapshot) =
dataset(out, job_ns);
let output_table = output_table.ok_or("output dataset has no `name`")?;
let base = LineageRow {
event_id,
actor,
operation,
input_table: None,
input_snapshot: None,
output_table,
output_snapshot,
source_system: None,
source_instance: None,
target_system,
target_instance,
ts_micros,
commit_seq,
};
let inputs = obj.get("inputs").and_then(|v| v.as_array());
match inputs {
Some(arr) if !arr.is_empty() => Ok(arr
.iter()
.map(|ds| {
let (in_table, sys, inst, snap) = dataset(ds, job_ns);
let mut r = base.clone();
r.input_table = in_table;
r.input_snapshot = snap;
r.source_system = sys;
r.source_instance = inst;
r
})
.collect()),
// No inputs: the output node still contributes (external ingest / create).
_ => Ok(vec![base]),
}
}
/// Parse a batch of OpenLineage `RunEvent`s — either a top-level JSON array of
/// events, or a single event object — into a flat [`LineageRow`] list.
///
/// # Errors
/// Propagates the first per-event decode error.
pub fn read_run_events(value: &serde_json::Value) -> Result<Vec<LineageRow>, String> {
let mut out = Vec::new();
match value {
serde_json::Value::Array(events) => {
for ev in events {
out.extend(read_run_event(ev)?);
}
}
other => out.extend(read_run_event(other)?),
}
Ok(out)
}
/// Decode one OpenLineage dataset `{namespace,name,facets}` into
/// `(table, system_kind, system_instance, snapshot)`. A namespace equal to the
/// job namespace is in-warehouse (`system = None`); any other namespace is a
/// foreign system whose *instance* is that namespace.
fn dataset(
ds: &serde_json::Value,
job_ns: &str,
) -> (Option<String>, Option<String>, Option<String>, Option<i64>) {
let name = ds.get("name").and_then(|v| v.as_str()).map(str::to_string);
let ns = ds.get("namespace").and_then(|v| v.as_str()).unwrap_or("");
let snapshot = ds
.get("facets")
.and_then(|f| f.get("version"))
.and_then(|ver| ver.get("datasetVersion"))
.and_then(|v| v.as_str().and_then(|s| s.parse::<i64>().ok()).or_else(|| v.as_i64()));
let (kind, inst) = if ns.is_empty() || ns == job_ns {
(None, None)
} else {
(Some("external".to_string()), Some(ns.to_string()))
};
(name, kind, inst, snapshot)
}
/// Recover skade's operation verb from a `"skade <op> on <table>"` documentation
/// description; `None` if it doesn't match that shape.
fn parse_operation(desc: &str) -> Option<String> {
let rest = desc.strip_prefix("skade ")?;
let op = rest.split(' ').next()?;
matches!(op, "append" | "delete" | "overwrite" | "create" | "release")
.then(|| op.to_string())
}
/// Parse an RFC-3339 / ISO-8601 UTC timestamp (`YYYY-MM-DDThh:mm:ss[.ffffff]Z`,
/// the exact shape skade's emitter writes) into epoch-microseconds. Pure date
/// math (Howard Hinnant's days-from-civil), no time-zone crate. `None` if the
/// prefix isn't a well-formed `date T time`.
#[must_use]
pub fn rfc3339_to_micros(s: &str) -> Option<i64> {
let bytes = s.as_bytes();
if bytes.len() < 19 || bytes[4] != b'-' || bytes[7] != b'-' || bytes[10] != b'T' {
return None;
}
let num = |lo: usize, hi: usize| -> Option<i64> { s.get(lo..hi)?.parse::<i64>().ok() };
let (y, mo, d) = (num(0, 4)?, num(5, 7)?, num(8, 10)?);
let (hh, mm, ss) = (num(11, 13)?, num(14, 16)?, num(17, 19)?);
// Fractional seconds (optional): ".ffffff" up to microsecond precision.
let mut micros_frac = 0i64;
if bytes.get(19) == Some(&b'.') {
let mut scale = 100_000i64; // first frac digit = 1e5 µs
for &b in &bytes[20..] {
if b.is_ascii_digit() && scale >= 1 {
micros_frac += i64::from(b - b'0') * scale;
scale /= 10;
} else {
break;
}
}
}
// days-from-civil (Hinnant): y/m/d → days since 1970-01-01.
let yy = if mo <= 2 { y - 1 } else { y };
let era = yy.div_euclid(400);
let yoe = yy - era * 400;
let mp = if mo > 2 { mo - 3 } else { mo + 9 };
let doy = (153 * mp + 2) / 5 + d - 1;
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
let days = era * 146_097 + doe - 719_468;
let secs = days * 86_400 + hh * 3_600 + mm * 60 + ss;
Some(secs * 1_000_000 + micros_frac)
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
/// A RunEvent shaped EXACTLY like skade's `HttpLineageSink::run_event`
/// output (cross-system transform hop: foreign `main.silver` → skade
/// `main.gold`) reads back into the right `LineageRow`.
#[test]
fn skade_run_event_reads_back_into_a_lineage_row() {
let ev = json!({
"eventType": "COMPLETE",
"eventTime": "2026-07-09T12:00:00.000512Z",
"producer": "https://git.gunnar.rs/nordisk/skade",
"run": {
"runId": "018f0000-0000-4000-8000-000000000abc",
"facets": { "skade_commit": { "commitSeq": 9001 } }
},
"job": {
"namespace": "lakehouse",
"name": "nornir/etl-run-3",
"facets": { "documentation": { "description": "skade append on main.gold" } }
},
"inputs": [{
"namespace": "app-42",
"name": "main.silver",
"facets": { "version": { "datasetVersion": "76" } }
}],
"outputs": [{
"namespace": "lakehouse",
"name": "main.gold",
"facets": { "version": { "datasetVersion": "77" } }
}]
});
let rows = read_run_event(&ev).expect("decode");
assert_eq!(rows.len(), 1, "one input → one row");
let r = &rows[0];
assert_eq!(r.event_id, "018f0000-0000-4000-8000-000000000abc");
assert_eq!(r.actor, "nornir/etl-run-3");
assert_eq!(r.operation, "append");
// Output is in the job namespace → in-warehouse (no target system).
assert_eq!(r.output_table, "main.gold");
assert_eq!(r.output_snapshot, Some(77));
assert_eq!(r.target_system, None);
// Input lives in a foreign namespace → foreign system, instance = ns.
assert_eq!(r.input_table.as_deref(), Some("main.silver"));
assert_eq!(r.input_snapshot, Some(76));
assert_eq!(r.source_system.as_deref(), Some("external"));
assert_eq!(r.source_instance.as_deref(), Some("app-42"));
assert_eq!(r.commit_seq, Some(9001));
// eventTime parsed to epoch-micros (…512 µs fractional preserved).
assert_eq!(r.ts_micros, rfc3339_to_micros("2026-07-09T12:00:00.000512Z").unwrap());
assert!(r.ts_micros > 1_700_000_000_000_000);
}
/// An input-less event (external ingest / table birth) still yields one row
/// carrying just the output node.
#[test]
fn input_less_event_yields_output_only_row() {
let ev = json!({
"eventTime": "2026-07-09T00:00:00Z",
"run": { "runId": "r1" },
"job": { "namespace": "skade", "name": "nornir/etl" },
"inputs": [],
"outputs": [{ "namespace": "skade", "name": "main.raw" }]
});
let rows = read_run_event(&ev).unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].output_table, "main.raw");
assert_eq!(rows[0].input_table, None);
assert_eq!(rows[0].operation, "append", "default op when no documentation facet");
}
/// A top-level JSON array of events flattens; a fan-in event with two inputs
/// becomes two rows (matching the reserved-table one-row-per-input shape).
#[test]
fn array_and_multi_input_fan_out() {
let events = json!([
{
"eventTime": "2026-07-09T00:00:00Z",
"run": { "runId": "a" },
"job": { "namespace": "skade", "name": "etl" },
"outputs": [{ "namespace": "skade", "name": "main.gold" }],
"inputs": [
{ "namespace": "skade", "name": "main.a" },
{ "namespace": "skade", "name": "main.b" }
]
},
{
"eventTime": "2026-07-09T00:00:01Z",
"run": { "runId": "b" },
"job": { "namespace": "skade", "name": "etl" },
"outputs": [{ "namespace": "skade", "name": "main.raw" }]
}
]);
let rows = read_run_events(&events).unwrap();
// 2 inputs (event a) + 1 input-less (event b) = 3 rows.
assert_eq!(rows.len(), 3);
let a_rows: Vec<_> = rows.iter().filter(|r| r.event_id == "a").collect();
assert_eq!(a_rows.len(), 2, "fan-in → one row per input");
assert!(a_rows.iter().any(|r| r.input_table.as_deref() == Some("main.a")));
assert!(a_rows.iter().any(|r| r.input_table.as_deref() == Some("main.b")));
}
/// The pure RFC-3339 → micros parser round-trips a known instant and handles
/// the no-fractional form.
#[test]
fn rfc3339_parses_epoch_micros() {
// 1970-01-01T00:00:00Z == 0.
assert_eq!(rfc3339_to_micros("1970-01-01T00:00:00Z"), Some(0));
// One day later.
assert_eq!(rfc3339_to_micros("1970-01-02T00:00:00Z"), Some(86_400_000_000));
// Fractional micros preserved.
assert_eq!(rfc3339_to_micros("1970-01-01T00:00:00.000001Z"), Some(1));
// Malformed → None.
assert_eq!(rfc3339_to_micros("not-a-time"), None);
}
}