Decompose problems into pipelines of data transformations. Refactors loops into map/filter/reduce chains, converts nested/OO logic into composable function sequences, designs multi-step data transformation pipelines. Trigger on: "transformational programming", "data pipeline", "function pipeline", "pipe operator", "|>", "stream processing", "chained transformations", "Unix pipes", "dataflow", "decompose into steps", "write this as a pipeline", "compose functions", "chain of transformations", or restructuring imperative/OO code into data transforms. NOT for ETL infrastructure or stream processing frameworks (Kafka, Flink) — focuses on code-level function composition and transformation design patterns.
94
97%
Does it follow best practices?
Impact
85%
1.19xAverage score across 3 eval scenarios
Passed
No known issues
Express the program as input → output, then decompose it into single-responsibility pipeline steps.
Validation checkpoint: each step has one input type and one output type; no step requires knowledge of another step's internal state.
Keep transformations input-only/output-only. Use language patterns for pipeline syntax and error handling per language.
Chain steps with pipe operators, method chaining, or explicit assignments:
# Explicit assignment chain (works in any language)
content = read_file(path)
lines = find_matching(content, pattern)
result = truncate(lines)
return sort_by_date(result)Before composing, verify each transformation works correctly with sample inputs. After composing, test the full pipeline end-to-end.
Wrap results in a success/failure type so errors propagate without breaking the chain. Each step either transforms a valid value or passes through an error unchanged. Use language patterns for language-specific patterns.
User: "I need a program that reads a log file, finds ERROR entries, extracts timestamp+message, and sorts by timestamp."
## Transformation Analysis: Error Log Extractor
### Data Flow
file path → file content → error lines → parsed entries → sorted entries
### Implementation (Python)
def read_file(path):
with open(path) as f:
return f.read()
def filter_errors(content):
return [line for line in content.split('\n') if 'ERROR' in line]
def parse_entries(lines):
return [
{'timestamp': parts[0], 'message': parts[2]}
for line in lines
if len(parts := line.split(' ', 2)) >= 3
]
def sort_by_timestamp(entries):
return sorted(entries, key=lambda e: e['timestamp'])
def extract_errors(path):
content = read_file(path) # 1
lines = filter_errors(content) # 2
entries = parse_entries(lines) # 3
return sort_by_timestamp(entries) # 4
### Error handling
- read_file raises FileNotFoundError for missing files
- parse_entries skips malformed lines
- Steps 2 and 4 are pure and can't fail