Function evolve_schema

nodedb/src/engine/timeseries/ilp_schema.rs:91–124

Detect new fields in an ILP batch and expand the memtable schema. Scans all lines for tag keys and field keys not present in the current schema. New columns are added with NULL backfill for existing rows. Must be called BEFORE `ingest_batch` so the batch can map values to the expanded schema.

(memtable: &mut ColumnarMemtable, lines: &[IlpLine<'_>])

/// Must be called BEFORE `ingest_batch` so the batch can map values to
/// the expanded schema.
pub fn evolve_schema(memtable: &mut ColumnarMemtable, lines: &[IlpLine<'_>]) {
    let existing: std::collections::HashSet<String> = memtable
        .schema()
        .columns
        .iter()
        .map(|(n, _)| n.clone())
        .collect();

    let mut new_columns: Vec<(String, ColumnType)> = Vec::new();
    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();

    for line in lines {
        for &(key, _) in &line.tags {
            if !existing.contains(key) && seen.insert(key.to_string()) {
                new_columns.push((key.to_string(), ColumnType::Symbol));
            }
        }
        for &(key, ref val) in &line.fields {
            if !existing.contains(key) && seen.insert(key.to_string()) {
                let col_type = match val {
                    FieldValue::Float(_) => ColumnType::Float64,
                    FieldValue::Int(_) | FieldValue::UInt(_) => ColumnType::Int64,
                    FieldValue::Str(_) => ColumnType::Symbol,
                    FieldValue::Bool(_) => ColumnType::Float64,
                };
                new_columns.push((key.to_string(), col_type));
            }
        }
    }

    for (name, col_type) in new_columns {
        memtable.add_column(name, col_type);
    }
}