use async_trait::async_trait;
use datafusion::prelude::*;
use serde::{Deserialize, Serialize};
use tracing::instrument;
use crate::analyzers::{Analyzer, AnalyzerError, AnalyzerResult, AnalyzerState, MetricValue};
use crate::core::current_validation_context;
#[derive(Debug, Clone)]
pub struct CompletenessAnalyzer {
column: String,
}
impl CompletenessAnalyzer {
pub fn new(column: impl Into<String>) -> Self {
Self {
column: column.into(),
}
}
pub fn column(&self) -> &str {
&self.column
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CompletenessState {
pub total_count: u64,
pub non_null_count: u64,
}
impl CompletenessState {
pub fn completeness(&self) -> f64 {
if self.total_count == 0 {
1.0 } else {
self.non_null_count as f64 / self.total_count as f64
}
}
}
impl AnalyzerState for CompletenessState {
fn merge(states: Vec<Self>) -> AnalyzerResult<Self> {
let total_count = states.iter().map(|s| s.total_count).sum();
let non_null_count = states.iter().map(|s| s.non_null_count).sum();
Ok(CompletenessState {
total_count,
non_null_count,
})
}
fn is_empty(&self) -> bool {
self.total_count == 0
}
}
#[async_trait]
impl Analyzer for CompletenessAnalyzer {
type State = CompletenessState;
type Metric = MetricValue;
#[instrument(skip(ctx), fields(analyzer = "completeness", column = %self.column))]
async fn compute_state_from_data(&self, ctx: &SessionContext) -> AnalyzerResult<Self::State> {
let validation_ctx = current_validation_context();
let table_name = validation_ctx.table_name();
let sql = format!(
"SELECT COUNT(*) as total_count, COUNT({}) as non_null_count FROM {table_name}",
self.column
);
let df = ctx.sql(&sql).await?;
let batches = df.collect().await?;
let (total_count, non_null_count) = if let Some(batch) = batches.first() {
if batch.num_rows() > 0 {
let total_array = batch
.column(0)
.as_any()
.downcast_ref::<arrow::array::Int64Array>()
.ok_or_else(|| {
AnalyzerError::invalid_data("Expected Int64 array for total count")
})?;
let non_null_array = batch
.column(1)
.as_any()
.downcast_ref::<arrow::array::Int64Array>()
.ok_or_else(|| {
AnalyzerError::invalid_data("Expected Int64 array for non-null count")
})?;
(total_array.value(0) as u64, non_null_array.value(0) as u64)
} else {
(0, 0)
}
} else {
(0, 0)
};
Ok(CompletenessState {
total_count,
non_null_count,
})
}
fn compute_metric_from_state(&self, state: &Self::State) -> AnalyzerResult<Self::Metric> {
Ok(MetricValue::Double(state.completeness()))
}
fn name(&self) -> &str {
"completeness"
}
fn description(&self) -> &str {
"Computes the fraction of non-null values in a column"
}
fn metric_key(&self) -> String {
format!("{}.{}", self.name(), self.column)
}
fn columns(&self) -> Vec<&str> {
vec![&self.column]
}
}