EnerOS v0.8.0

Release Date: August 4, 2024 Codename: TimeSeries Git Tag: v0.8.0 Support Status: Internal Preview Total Crates: 7 Test Cases: 1108

Version Overview

EnerOS v0.8.0 “TimeSeries” introduces the time series data storage engine. This version releases the eneros-timeseries crate, based on the Apache Arrow columnar storage format, providing high-throughput, low-latency grid measurement data storage and query capabilities. Power systems are naturally time-series-data-intensive scenarios — a provincial dispatch center receives hundreds of thousands of telemetry data points per second (voltage, current, power, frequency), with daily data volumes reaching TB level. Traditional relational databases cannot handle this write throughput and query pattern; EnerOS chose Apache Arrow as the underlying in-memory format, combined with Parquet persistence, achieving a “write is query” zero-copy pipeline.

The time series engine of v0.8.0 makes targeted optimizations for the data characteristics of power systems. First, grid measurement data has strong “temporal locality” — the same measurement point changes little over short time periods, so the engine defaults to Delta encoding compression, with typical compression ratios reaching 10:1. Second, grid analysis queries are usually “by measurement point + time range” retrieval, rather than “traverse all measurement points by time”, so the engine uses a “measurement-point-first” inverted index, allowing the offset of all data blocks for that measurement point to be located given a measurement point ID. Third, historical data access has a “recent-hot, distant-cold” characteristic — the most recent 1 hour of data is queried most frequently, 1 day ago is next, and 1 year ago is rarely accessed, so the engine implements tiered storage: hot data stays in memory, warm data is persisted to Parquet, and cold data can be archived to object storage.

This version also introduces a downsampling mechanism, supporting the aggregation of high-frequency raw data into low-frequency statistical data by time window. For example, raw waveform data with 1ms sampling interval can be downsampled to 1s average, maximum, minimum, and standard deviation, reducing storage space by 1000x while retaining key statistical characteristics. Downsampling rules are configurable: voltage/frequency is suitable for averages, fault current for maximums, and alarm counts for sums. The engine also supports historical data backfill — when field terminals lose data due to communication interruptions, they can batch-upload after connection is restored, and the engine will automatically insert data at the correct position by timestamp.

Key Data

New Features

1. Arrow-based Columnar Storage

// crates/eneros-timeseries/src/storage.rs
use arrow::array::{Float64Array, Int64Array, TimestampNanosecondArray};
use arrow::record_batch::RecordBatch;
use arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;

/// Time series data point
#[derive(Debug, Clone)]
pub struct DataPoint {
    pub timestamp_ns: i64,  // Nanosecond timestamp
    pub value: f64,
    pub quality: u8,        // Data quality flag
}

/// Time series storage engine
pub struct TimeSeriesEngine {
    /// Measurement point ID → data block index
    index: HashMap<MeasurementId, Vec<DataBlock>>,
    /// In-memory hot data buffer
    hot_buffer: HashMap<MeasurementId, VecDeque<DataPoint>>,
    /// Configuration
    config: EngineConfig,
}

#[derive(Debug, Clone)]
pub struct EngineConfig {
    /// Hot data retention duration (seconds)
    pub hot_retention_secs: u64,
    /// Maximum points per data block
    pub block_max_points: usize,
    /// Whether to enable compression
    pub compression: Compression,
    /// Downsampling rules
    pub downsampling: Vec<DownsampleRule>,
}

#[derive(Debug, Clone, Copy)]
pub enum Compression {
    None,
    Delta,
    Snappy,
    Zstd(i32),
}

impl TimeSeriesEngine {
    pub fn new(config: EngineConfig) -> Self {
        Self {
            index: HashMap::new(),
            hot_buffer: HashMap::new(),
            config,
        }
    }

    /// Write a single data point
    pub fn write(&mut self, id: MeasurementId, point: DataPoint) -> EnerOSResult<()> {
        let buffer = self.hot_buffer.entry(id).or_insert_with(VecDeque::new);
        buffer.push_back(point);

        // Flush to Arrow data block when buffer is full
        if buffer.len() >= self.config.block_max_points {
            self.flush_block(id)?;
        }
        Ok(())
    }

    /// Batch write
    pub fn write_batch(&mut self, id: MeasurementId, points: Vec<DataPoint>) -> EnerOSResult<()> {
        let buffer = self.hot_buffer.entry(id).or_insert_with(VecDeque::new);
        for p in points {
            buffer.push_back(p);
        }
        if buffer.len() >= self.config.block_max_points {
            self.flush_block(id)?;
        }
        Ok(())
    }

    /// Flush buffer data to Arrow RecordBatch
    fn flush_block(&mut self, id: MeasurementId) -> EnerOSResult<()> {
        let buffer = self.hot_buffer.get_mut(&id).unwrap();
        let n = buffer.len();

        let mut timestamps = Vec::with_capacity(n);
        let mut values = Vec::with_capacity(n);
        let mut qualities = Vec::with_capacity(n);

        while let Some(p) = buffer.pop_front() {
            timestamps.push(p.timestamp_ns);
            values.push(p.value);
            qualities.push(p.quality);
        }

        // Build Arrow RecordBatch
        let schema = Arc::new(Schema::new(vec![
            Field::new("timestamp", DataType::Timestamp(arrow::datatypes::TimeUnit::Nanosecond, None), false),
            Field::new("value", DataType::Float64, false),
            Field::new("quality", DataType::UInt8, false),
        ]));

        let batch = RecordBatch::try_new(schema, vec![
            Arc::new(TimestampNanosecondArray::from(timestamps)),
            Arc::new(Float64Array::from(values)),
            Arc::new(qualities.into_iter().map(|q| q as u8).collect::<arrow::array::UInt8Array>().into()),
        ])?;

        let block = DataBlock {
            batch,
            start_ts: timestamps.first().copied().unwrap_or(0),
            end_ts: timestamps.last().copied().unwrap_or(0),
            count: n,
        };

        self.index.entry(id).or_default().push(block);
        Ok(())
    }
}

2. Timestamp Index

// crates/eneros-timeseries/src/index.rs
use std::collections::BTreeMap;

/// Measurement point ID
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct MeasurementId(pub u64);

/// Time range index: quickly locate data blocks based on BTree
pub struct TimeIndex {
    /// Timestamp → data block offset
    blocks: BTreeMap<i64, BlockLocation>,
}

#[derive(Debug, Clone)]
pub struct BlockLocation {
    pub block_id: u64,
    pub start_ts: i64,
    pub end_ts: i64,
    pub point_count: usize,
}

impl TimeIndex {
    pub fn new() -> Self {
        Self { blocks: BTreeMap::new() }
    }

    /// Find data blocks covering the time range
    pub fn find_range(&self, start: i64, end: i64) -> Vec<&BlockLocation> {
        self.blocks.range(..=end)
            .filter(|(_, loc)| loc.start_ts >= start || loc.end_ts >= start)
            .map(|(_, loc)| loc)
            .collect()
    }
}

3. Downsampling Mechanism

// crates/eneros-timeseries/src/downsample.rs
use crate::DataPoint;

/// Downsampling aggregation function
#[derive(Debug, Clone, Copy)]
pub enum Aggregation {
    Avg,    // Average
    Min,    // Minimum
    Max,    // Maximum
    Sum,    // Sum
    Count,  // Count
    Stddev, // Standard deviation
    Last,   // Last value
}

/// Downsampling rule
#[derive(Debug, Clone)]
pub struct DownsampleRule {
    /// Original sampling interval (nanoseconds)
    pub source_interval_ns: i64,
    /// Target sampling interval (nanoseconds)
    pub target_interval_ns: i64,
    /// Aggregation function
    pub aggregation: Aggregation,
}

impl DownsampleRule {
    /// Execute downsampling
    pub fn downsample(&self, points: &[DataPoint]) -> Vec<DataPoint> {
        let window = self.target_interval_ns;
        let mut result = Vec::new();
        let mut window_start = points.first().map(|p| p.timestamp_ns).unwrap_or(0);

        let mut current_window: Vec<&DataPoint> = Vec::new();

        for p in points {
            if p.timestamp_ns - window_start >= window {
                if !current_window.is_empty() {
                    let agg_value = self.aggregate(&current_window);
                    result.push(DataPoint {
                        timestamp_ns: window_start,
                        value: agg_value,
                        quality: 0,
                    });
                }
                window_start = p.timestamp_ns;
                current_window.clear();
            }
            current_window.push(p);
        }

        if !current_window.is_empty() {
            let agg_value = self.aggregate(&current_window);
            result.push(DataPoint {
                timestamp_ns: window_start,
                value: agg_value,
                quality: 0,
            });
        }
        result
    }

    fn aggregate(&self, points: &[&DataPoint]) -> f64 {
        match self.aggregation {
            Aggregation::Avg => {
                let sum: f64 = points.iter().map(|p| p.value).sum();
                sum / points.len() as f64
            }
            Aggregation::Min => points.iter().map(|p| p.value).fold(f64::INFINITY, f64::min),
            Aggregation::Max => points.iter().map(|p| p.value).fold(f64::NEG_INFINITY, f64::max),
            Aggregation::Sum => points.iter().map(|p| p.value).sum(),
            Aggregation::Count => points.len() as f64,
            Aggregation::Stddev => {
                let mean: f64 = points.iter().map(|p| p.value).sum::<f64>() / points.len() as f64;
                let var: f64 = points.iter().map(|p| (p.value - mean).powi(2)).sum::<f64>() / points.len() as f64;
                var.sqrt()
            }
            Aggregation::Last => points.last().unwrap().value,
        }
    }
}

Downsampling example:

use eneros_timeseries::{TimeSeriesEngine, EngineConfig, DownsampleRule, Aggregation};

let mut engine = TimeSeriesEngine::new(EngineConfig {
    hot_retention_secs: 3600,
    block_max_points: 8192,
    compression: Compression::Delta,
    downsampling: vec![
        // 1ms raw → 1s average
        DownsampleRule {
            source_interval_ns: 1_000_000,
            target_interval_ns: 1_000_000_000,
            aggregation: Aggregation::Avg,
        },
        // 1s → 1min maximum
        DownsampleRule {
            source_interval_ns: 1_000_000_000,
            target_interval_ns: 60_000_000_000,
            aggregation: Aggregation::Max,
        },
    ],
});

// Write data
for i in 0..1000 {
    engine.write(measurement_id, DataPoint {
        timestamp_ns: i * 1_000_000,
        value: 220.0 + (i as f64 * 0.01).sin(),
        quality: 0,
    })?;
}

4. Historical Data Query

// crates/eneros-timeseries/src/query.rs
use crate::{MeasurementId, DataPoint, TimeSeriesEngine};

/// Query conditions
#[derive(Debug, Clone)]
pub struct Query {
    pub measurement_id: MeasurementId,
    pub start_ts: i64,
    pub end_ts: i64,
    /// Downsampling rule (optional)
    pub downsample: Option<DownsampleRule>,
    /// Quality filter
    pub min_quality: u8,
}

impl TimeSeriesEngine {
    /// Query historical data
    pub fn query(&self, q: Query) -> EnerOSResult<Vec<DataPoint>> {
        let blocks = self.index.get(&q.measurement_id)
            .ok_or(TimeSeriesError::MeasurementNotFound(q.measurement_id))?;

        let mut result = Vec::new();
        for block in blocks {
            // Skip data blocks not in the time range
            if block.end_ts < q.start_ts || block.start_ts > q.end_ts {
                continue;
            }

            // Read data from Arrow RecordBatch
            let timestamps = block.batch.column(0)
                .as_any().downcast_ref::<TimestampNanosecondArray>().unwrap();
            let values = block.batch.column(1)
                .as_any().downcast_ref::<Float64Array>().unwrap();
            let qualities = block.batch.column(2)
                .as_any().downcast_ref::<UInt8Array>().unwrap();

            for i in 0..block.count {
                let ts = timestamps.value(i);
                if ts < q.start_ts || ts > q.end_ts {
                    continue;
                }
                if qualities.value(i) < q.min_quality {
                    continue;
                }
                result.push(DataPoint {
                    timestamp_ns: ts,
                    value: values.value(i),
                    quality: qualities.value(i),
                });
            }
        }

        // Apply downsampling
        if let Some(rule) = q.downsample {
            result = rule.downsample(&result);
        }

        Ok(result)
    }
}

Improvements

• eneros-core: Added MeasurementId type
• eneros-agent: Agents can access time series data via syscalls
• Dependencies: Introduced arrow, parquet

Bug Fixes

• Fixed flush_block panicking on empty buffer (#115)
• Fixed query missing data when time range spans data blocks (#119)
• Fixed DownsampleRule::downsample not outputting the last window (#123)

Breaking Changes

• eneros_core: Added MeasurementId type; original u64 measurement point IDs need wrapping

Performance Improvements

Compression ratio comparison:

Contributors

Upgrade Guide

New Dependencies

[dependencies]
eneros-timeseries = { version = "0.8", path = "../eneros-timeseries" }
arrow = "48"
parquet = "48"

Basic Usage

use eneros_timeseries::{TimeSeriesEngine, EngineConfig, Compression, DataPoint};

let mut engine = TimeSeriesEngine::new(EngineConfig {
    hot_retention_secs: 3600,
    block_max_points: 8192,
    compression: Compression::Delta,
    downsampling: vec![],
});

let id = MeasurementId(1001);
engine.write(id, DataPoint {
    timestamp_ns: chrono::Utc::now().timestamp_nanos(),
    value: 220.5,
    quality: 0,
})?;