datafusion_openlineage/
config.rs

1//! DataFusion engine identity for [`OpenLineageConfig`].
2//!
3//! [`OpenLineageConfig`] itself is engine-agnostic and lives in
4//! [`openlineage_client`]. This module adds [`DataFusionConfig`], an extension
5//! trait that stamps the DataFusion `processing_engine` identity (name +
6//! [`datafusion::DATAFUSION_VERSION`]) and a DataFusion-flavored producer.
7
8pub use openlineage_client::OpenLineageConfig;
9
10/// The `producer` URI stamped on events emitted by this integration.
11pub const DATAFUSION_PRODUCER: &str =
12    "https://github.com/open-lakehouse/headwaters/datafusion-openlineage";
13
14/// Builds [`OpenLineageConfig`] with the DataFusion `processing_engine` identity.
15pub trait DataFusionConfig {
16    /// A config with the DataFusion producer/engine identity and default
17    /// namespace.
18    fn for_datafusion() -> Self;
19
20    /// [`Self::for_datafusion`] but reading `OPENLINEAGE_NAMESPACE` /
21    /// `OPENLINEAGE_TIMEOUT_MS` from the environment for the namespace and
22    /// transport timeout (the engine identity is still fixed).
23    fn for_datafusion_from_env() -> Self;
24}
25
26impl DataFusionConfig for OpenLineageConfig {
27    fn for_datafusion() -> Self {
28        stamp_datafusion(OpenLineageConfig {
29            producer: DATAFUSION_PRODUCER.to_string(),
30            ..OpenLineageConfig::with_env(None, None)
31        })
32    }
33
34    fn for_datafusion_from_env() -> Self {
35        stamp_datafusion(OpenLineageConfig {
36            producer: DATAFUSION_PRODUCER.to_string(),
37            ..OpenLineageConfig::from_env()
38        })
39    }
40}
41
42/// Fill in the DataFusion `processing_engine` identity: this integration is the
43/// adapter, DataFusion is the engine, so report DataFusion's version as the
44/// engine version (the crate version stays the adapter version).
45fn stamp_datafusion(mut cfg: OpenLineageConfig) -> OpenLineageConfig {
46    cfg.engine_name = "DataFusion".to_string();
47    cfg.engine_version = datafusion::DATAFUSION_VERSION.to_string();
48    cfg
49}
50
51#[cfg(test)]
52mod tests {
53    use super::*;
54
55    #[test]
56    fn datafusion_identity_is_stable() {
57        let cfg = OpenLineageConfig::for_datafusion();
58        assert_eq!(cfg.producer, DATAFUSION_PRODUCER);
59        assert_eq!(cfg.engine_name, "DataFusion");
60        assert_eq!(cfg.adapter_version, env!("CARGO_PKG_VERSION"));
61        assert!(!cfg.engine_version.is_empty());
62    }
63}
datafusion_openlineage/config.rs

datafusion_openlineage/
config.rs
