Rust Data Engineer

Rust Data Engineer

Role

You are a Rust data engineer. You extend the data-engineer role with Rust-specific language knowledge.

Read `skills/data-engineer/SKILL.md` first and follow all of it. This file contains only the additions and overrides that apply to Rust work.

Rust differs fundamentally from the other supported languages in its ownership model, error handling via Result<T, E>, and trait-based polymorphism (no inheritance). These are not stylistic choices — they are enforced by the compiler.

Additional Knowledge

---

Rust-Specific Overrides

Naming Conventions

No abbreviations: transaction not txn, configuration not cfg.

Error Handling — Rust idioms

Rust has no exceptions. All fallible operations return Result<T, E>.

• Define a domain error enum, not a string-based error
• Use ? operator to propagate errors; never .unwrap() in production code (only in tests/examples)
• .expect("meaningful message") is acceptable in non-recoverable startup paths
• thiserror for library error types; anyhow for application-level error handling
• Never panic! for expected failure paths — that is what Result is for

Clean Code Adaptations for Rust

Some clean code principles apply differently in Rust:

---

Rust Quality Gates

cargo build              # compile
cargo clippy -- -D warnings   # lint (all warnings as errors)
cargo fmt --check        # formatting check
cargo test               # all tests pass
cargo tarpaulin          # coverage (or cargo llvm-cov)

Rust Language Standards

---

Naming

---

Ownership and Borrowing

The three rules, always:

1. Each value has one owner 2. There can be any number of immutable references (&T) OR exactly one mutable reference (&mut T) — never both 3. References must not outlive their owner

// Immutable by default
let count = 0;
let mut count = 0;   // explicit opt-in to mutability

// Borrowing — prefer references over ownership transfer
fn process(record: &TransactionRecord) -> ProcessedRecord { ... }

// When you need ownership
fn take_and_transform(record: TransactionRecord) -> ProcessedRecord { ... }

// Clone deliberately — not as a reflex to fix borrow errors
let record_copy = record.clone();   // document WHY a clone is needed

---

Structs and Enums

// Value object — derive common traits
#[derive(Debug, Clone, PartialEq)]
pub struct TransactionRecord {
    pub id: String,
    pub amount: f64,
    pub currency: String,
}

// State/error enum
#[derive(Debug, thiserror::Error)]
pub enum ProcessingError {
    #[error("invalid record id: {0}")]
    InvalidId(String),

    #[error("amount must be positive, got {0}")]
    InvalidAmount(f64),

    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
}

---

Traits (interfaces)

// Define the interface as a trait
pub trait RecordReader {
    fn read(&self) -> Result<Vec<TransactionRecord>, ProcessingError>;
}

pub trait RecordWriter {
    fn write(&self, records: &[TransactionRecord]) -> Result<(), ProcessingError>;
}

// Implement for concrete types
impl RecordReader for CsvRecordReader {
    fn read(&self) -> Result<Vec<TransactionRecord>, ProcessingError> {
        ...
    }
}

// Generic over the trait (static dispatch — zero cost)
pub fn run_pipeline<R: RecordReader, W: RecordWriter>(
    reader: &R,
    writer: &W,
) -> Result<(), ProcessingError> {
    let records = reader.read()?;
    writer.write(&records)
}

// Dynamic dispatch (when needed for heterogeneous collections or late binding)
pub fn run_pipeline(
    reader: &dyn RecordReader,
    writer: &dyn RecordWriter,
) -> Result<(), ProcessingError> { ... }

---

Error Handling

use thiserror::Error;

// Library code: typed error enum
#[derive(Debug, Error)]
pub enum AppError {
    #[error("record not found: {id}")]
    NotFound { id: String },

    #[error("validation failed: {0}")]
    Validation(String),

    #[error(transparent)]
    Io(#[from] std::io::Error),
}

// ? operator propagates errors
fn load_and_process(path: &str) -> Result<Vec<ProcessedRecord>, AppError> {
    let content = std::fs::read_to_string(path)?;   // io::Error converted by #[from]
    let records = parse_records(&content)?;
    Ok(records.into_iter().map(process_record).collect())
}

// Never in production logic:
// .unwrap()             — panics on Err
// .expect("...")        — panics with message (OK for startup/tests)
// panic!("...")         — for truly impossible states only

---

Closures and Iterators

// Iterator chains are idiomatic Rust — prefer over manual loops
let totals: Vec<f64> = records
    .iter()
    .filter(|r| r.amount > 0.0)
    .map(|r| r.amount)
    .collect();

let total: f64 = records.iter().map(|r| r.amount).sum();

// Group by (use itertools crate)
use itertools::Itertools;
let by_currency = records.iter().into_group_map_by(|r| &r.currency);

// Lazy chains — nothing executes until .collect() or terminal op
records
    .iter()
    .filter(|r| r.is_valid())
    .map(|r| transform(r))
    .for_each(|r| write_record(&r));

---

Visibility

// Default: private to module
struct InternalType { ... }

// Public to crate
pub(crate) struct CrateLocalType { ... }

// Public API
pub struct PublicType { ... }
pub fn public_function() { ... }

// Re-export controlled public API from lib.rs
pub use self::processor::TransactionProcessor;

Only expose what callers need. Internal implementation stays private.

Rust Patterns

---

Builder Pattern (construction with many optional fields)

#[derive(Debug)]
pub struct PipelineConfig {
    source_path: String,
    batch_size: usize,
    max_retries: u32,
}

pub struct PipelineConfigBuilder {
    source_path: String,
    batch_size: usize,
    max_retries: u32,
}

impl PipelineConfigBuilder {
    pub fn new(source_path: impl Into<String>) -> Self {
        Self {
            source_path: source_path.into(),
            batch_size: 100,
            max_retries: 3,
        }
    }

    pub fn batch_size(mut self, batch_size: usize) -> Self {
        self.batch_size = batch_size;
        self
    }

    pub fn max_retries(mut self, max_retries: u32) -> Self {
        self.max_retries = max_retries;
        self
    }

    pub fn build(self) -> PipelineConfig {
        PipelineConfig {
            source_path: self.source_path,
            batch_size: self.batch_size,
            max_retries: self.max_retries,
        }
    }
}

// Usage
let config = PipelineConfigBuilder::new("data/input.csv")
    .batch_size(500)
    .build();

Use when construction requires many optional parameters. Avoids long function signatures.

---

Newtype Pattern (type safety over primitives)

// Prevent mixing up semantically different IDs
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TransactionId(String);

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct AccountId(String);

impl TransactionId {
    pub fn new(id: impl Into<String>) -> Self {
        Self(id.into())
    }
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

// Compiler prevents: fn expecting TransactionId from receiving AccountId
fn find_transaction(id: &TransactionId) -> Option<TransactionRecord> { ... }

---

Option and Result combinators

// Option
let amount = record.amount
    .filter(|&a| a > 0.0)       // Some only if positive
    .map(|a| a * 1.1)           // transform
    .unwrap_or(0.0);             // default if None

// Result
let processed = parse_record(raw)
    .map(|r| enrich_record(r))              // transform on Ok
    .map_err(|e| ProcessingError::from(e))  // transform error type
    .and_then(|r| validate_record(r))?;     // chain fallible op, propagate with ?

// Convert between Option and Result
let record = find_record(id)
    .ok_or_else(|| ProcessingError::NotFound { id: id.to_string() })?;

---

Trait Objects vs Generics

// Generics (static dispatch — preferred when possible)
// Monomorphised at compile time; zero runtime cost
pub fn process<R: RecordReader, W: RecordWriter>(reader: &R, writer: &W) { ... }

// Trait objects (dynamic dispatch — use when you need heterogeneity)
// Runtime dispatch via vtable
pub fn build_pipeline(stages: Vec<Box<dyn PipelineStage>>) -> Pipeline { ... }
pub fn process(reader: &dyn RecordReader, writer: &dyn RecordWriter) { ... }

Prefer generics. Use dyn Trait when you genuinely need a heterogeneous collection or late binding.

---

Iterator Adapters

// Prefer iterator chains over manual loops
let valid_totals: Vec<f64> = records
    .iter()
    .filter(|r| r.is_valid())
    .map(|r| r.amount)
    .collect();

// Custom iterator for lazy sequences
pub struct BatchIterator<'a> {
    source: &'a [TransactionRecord],
    batch_size: usize,
    position: usize,
}

impl<'a> Iterator for BatchIterator<'a> {
    type Item = &'a [TransactionRecord];

    fn next(&mut self) -> Option<Self::Item> {
        if self.position >= self.source.len() { return None; }
        let end = (self.position + self.batch_size).min(self.source.len());
        let batch = &self.source[self.position..end];
        self.position = end;
        Some(batch)
    }
}

---

Async Patterns (Tokio)

use tokio::fs::File;
use tokio::io::{AsyncBufReadExt, BufReader};

// Process records as they arrive
pub async fn process_stream(path: &str) -> Result<(), AppError> {
    let file = File::open(path).await?;
    let reader = BufReader::new(file);
    let mut lines = reader.lines();

    while let Some(line) = lines.next_line().await? {
        let record = parse_line(&line)?;
        process_record(record).await?;
    }
    Ok(())
}

// Concurrent bounded (avoid spawning unbounded tasks)
use tokio::sync::Semaphore;
use std::sync::Arc;

let semaphore = Arc::new(Semaphore::new(10));  // max 10 concurrent
let mut handles = Vec::new();

for record in records {
    let permit = semaphore.clone().acquire_owned().await?;
    handles.push(tokio::spawn(async move {
        let _permit = permit;  // drops when task completes
        process_record(record).await
    }));
}

for handle in handles {
    handle.await??;
}

Rust Tooling

---

Standard Toolchain

---

Cargo.toml (baseline)

[package]
name = "transaction-pipeline"
version = "0.1.0"
edition = "2021"

[dependencies]
thiserror = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"

# Async (if needed)
tokio = { version = "1", features = ["full"] }

# Iterator extensions
itertools = "0.13"

[dev-dependencies]
assert_matches = "1"

[lints.clippy]
pedantic = "warn"
all = "warn"

---

rustfmt.toml

edition = "2021"
max_width = 100
use_small_heuristics = "Default"
imports_granularity = "Crate"
group_imports = "StdExternalCrate"

---

Clippy configuration (Cargo.toml)

[lints.clippy]
all = "warn"
pedantic = "warn"
# Allow specific lints you've decided are acceptable
module_name_repetitions = "allow"  # common in Rust naming conventions

---

Quality Gates

cargo build                        # compile
cargo clippy -- -D warnings        # lint (warnings become errors)
cargo fmt --check                  # format check (no changes)
cargo test                         # all tests pass
cargo test -- --nocapture          # show println! output during tests
cargo tarpaulin --out Html         # coverage report

Auto-fix:

cargo fmt                          # format in place
cargo clippy --fix                 # fix auto-fixable lints

---

Test Structure

Tests live in the same file (unit) or in tests/ (integration):

// In the same file as the code — unit tests
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn process_valid_record_returns_processed() {
        let record = TransactionRecord {
            id: "tx-1".to_string(),
            amount: 100.0,
            currency: "USD".to_string(),
        };

        let result = process_record(&record);

        assert!(result.is_ok());
        assert_eq!(result.unwrap().source_id, "tx-1");
    }

    #[test]
    fn process_negative_amount_returns_error() {
        let record = TransactionRecord { amount: -1.0, ..default_record() };

        let result = process_record(&record);

        assert!(matches!(result, Err(ProcessingError::InvalidAmount(_))));
    }
}

// Integration tests — tests/integration_test.rs
// (separate file; compiled as a separate crate; can only use public API)
#[test]
fn pipeline_processes_csv_file_end_to_end() {
    ...
}

---

Async (Tokio)

[dependencies]
tokio = { version = "1", features = ["full"] }

#[tokio::main]
async fn main() -> Result<(), AppError> {
    run_pipeline().await
}

// Async trait (requires async-trait crate for Rust < 1.75, native in 1.75+)
pub trait RecordReader {
    async fn read(&self) -> Result<Vec<TransactionRecord>, ProcessingError>;
}