openlineage_client/
context.rs

1//! Top-level orchestration context contributed per emitted run.
2//!
3//! Different orchestration systems (Airflow, Dagster, Databricks Jobs) model
4//! run/job identity differently. Rather than hardcode a field set, an emitting
5//! integration contributes a [`LineageContext`] per query. The mechanism by
6//! which a context is produced is engine-specific (e.g. DataFusion derives it
7//! from a `SessionState`), so the *provider* trait lives with the engine
8//! integration — this crate owns only the context data and its env conventions.
9
10use serde_json::{Map, Value};
11use uuid::Uuid;
12
13use crate::config::OpenLineageConfig;
14use crate::facets::{BaseFacet, ParentJob, ParentRun, ParentRunFacet};
15
16/// Top-level context an integration contributes to each emitted run.
17///
18/// All fields are optional; unset values fall back to [`OpenLineageConfig`]
19/// defaults and plan-derived identity.
20#[derive(Debug, Default, Clone)]
21pub struct LineageContext {
22    /// Correlate with an orchestrator-owned run id (else a fresh UUIDv7 is used).
23    pub run_id: Option<Uuid>,
24    /// Namespace of the emitting job (else the configured default namespace).
25    pub job_namespace: Option<String>,
26    /// Name of the emitting job (else a plan-derived job name).
27    pub job_name: Option<String>,
28    /// Standard OpenLineage `parent` run facet.
29    pub parent_run: Option<ParentRunFacet>,
30    /// Arbitrary extra run facets merged into the emitted event.
31    pub run_facets: Map<String, Value>,
32    /// Arbitrary extra job facets merged into the emitted event.
33    pub job_facets: Map<String, Value>,
34    /// The SQL text of the query, if the host has it. Populates the `sql` job
35    /// facet. A plan walk cannot recover this, so the integration supplies it
36    /// from the request boundary.
37    pub sql: Option<String>,
38}
39
40impl LineageContext {
41    /// Build a context from the established OpenLineage parent-run environment
42    /// conventions, returning `None`-filled fields when nothing is set.
43    ///
44    /// Reads `OPENLINEAGE_PARENT_ID` (slash form `{namespace}/{name}/{runId}`),
45    /// falling back to the discrete `OPENLINEAGE_PARENT_JOB_NAMESPACE` /
46    /// `OPENLINEAGE_PARENT_JOB_NAME` / `OPENLINEAGE_PARENT_RUN_ID` variables.
47    pub fn from_env(config: &OpenLineageConfig) -> Self {
48        let parent_run = parent_from_env(config);
49        LineageContext {
50            parent_run,
51            ..Default::default()
52        }
53    }
54}
55
56fn parent_from_env(config: &OpenLineageConfig) -> Option<ParentRunFacet> {
57    let (namespace, name, run_id) = if let Ok(parent_id) = std::env::var("OPENLINEAGE_PARENT_ID") {
58        // Format: {namespace}/{name}/{runId}
59        let parts: Vec<&str> = parent_id.splitn(3, '/').collect();
60        match parts.as_slice() {
61            [ns, n, rid] => (ns.to_string(), n.to_string(), rid.to_string()),
62            _ => return None,
63        }
64    } else {
65        let ns = std::env::var("OPENLINEAGE_PARENT_JOB_NAMESPACE").ok()?;
66        let n = std::env::var("OPENLINEAGE_PARENT_JOB_NAME").ok()?;
67        let rid = std::env::var("OPENLINEAGE_PARENT_RUN_ID").ok()?;
68        (ns, n, rid)
69    };
70
71    Some(ParentRunFacet {
72        base: BaseFacet::new(&config.producer, "1-1-0/ParentRunFacet.json"),
73        run: ParentRun { run_id },
74        job: ParentJob { namespace, name },
75        root: None,
76    })
77}
openlineage_client/context.rs

openlineage_client/
context.rs
