Stage Category: REDUCE (Groups results by time)Transformation: N documents → M time-window results (with optional drift detection)
When to Use
| Use Case | Description |
|---|---|
| Trend analysis | Track how metrics change over time |
| Spike detection | Flag windows where a metric jumps or drops significantly |
| Content velocity | Count new documents per day/week/month |
| Temporal distribution | Understand when content was created or modified |
When NOT to Use
| Scenario | Recommended Alternative |
|---|---|
| Simple counting | aggregate |
| Full time-series from database | Aggregation API |
| Grouping by non-time fields | group_by |
| LLM-based trend analysis | summarize |
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
time_field | string | Required | Document field containing the timestamp |
window | string | Required | Window granularity: hour, day, week, month, quarter, year |
aggregations | array | Required | List of aggregation operations per window |
drift | object | null | Drift detection configuration |
sort_order | string | "asc" | Sort windows asc (oldest first) or desc (newest first) |
limit | integer | null | Max number of windows to return |
include_documents | boolean | false | Include original documents in output |
Aggregation Types
| Type | Field Required | Description |
|---|---|---|
count | No | Number of documents in window |
sum | Yes | Sum of field values |
avg | Yes | Average value |
min | Yes | Minimum value |
max | Yes | Maximum value |
count_distinct | Yes | Unique values count |
collect_distinct | Yes | Gather unique values into list |
Drift Detection
Drift detection compares a metric between consecutive windows and computes the percent change.| Parameter | Type | Default | Description |
|---|---|---|---|
drift.enabled | boolean | false | Enable drift detection |
drift.metric | string | Required if enabled | Which aggregation alias to track |
drift.threshold | float | null | Percent change to flag (e.g., 50.0 flags changes > 50%) |
Timestamp Formats
The stage parses timestamps automatically:| Format | Example |
|---|---|
| ISO 8601 | "2026-04-01T10:30:00Z" |
| ISO date | "2026-04-01" |
| Epoch seconds (int) | 1743465600 |
| Epoch seconds (float) | 1743465600.123 |
num_documents_skipped).
Configuration Examples
{
"stage_name": "temporal",
"stage_type": "reduce",
"config": {
"stage_id": "temporal",
"parameters": {
"time_field": "created_at",
"window": "day",
"aggregations": [
{"function": "count", "alias": "posts_per_day"},
{"function": "avg", "field": "score", "alias": "avg_relevance"}
]
}
}
}
{
"stage_name": "temporal",
"stage_type": "reduce",
"config": {
"stage_id": "temporal",
"parameters": {
"time_field": "published_at",
"window": "day",
"aggregations": [
{"function": "count", "alias": "count"}
],
"drift": {
"enabled": true,
"metric": "count",
"threshold": 50.0
}
}
}
}
{
"stage_name": "temporal",
"stage_type": "reduce",
"config": {
"stage_id": "temporal",
"parameters": {
"time_field": "created_at",
"window": "month",
"aggregations": [
{"function": "count", "alias": "total"},
{"function": "count_distinct", "field": "author", "alias": "unique_authors"},
{"function": "collect_distinct", "field": "category", "alias": "categories"}
],
"sort_order": "desc",
"limit": 12
}
}
}
{
"stage_name": "temporal",
"stage_type": "reduce",
"config": {
"stage_id": "temporal",
"parameters": {
"time_field": "transaction_date",
"window": "quarter",
"aggregations": [
{"function": "sum", "field": "amount", "alias": "revenue"},
{"function": "count", "alias": "transactions"},
{"function": "avg", "field": "amount", "alias": "avg_transaction"}
],
"drift": {
"enabled": true,
"metric": "revenue",
"threshold": 20.0
}
}
}
}
{
"stage_name": "temporal",
"stage_type": "reduce",
"config": {
"stage_id": "temporal",
"parameters": {
"time_field": "timestamp",
"window": "hour",
"aggregations": [
{"function": "count", "alias": "events"},
{"function": "max", "field": "latency_ms", "alias": "peak_latency"}
],
"drift": {
"enabled": true,
"metric": "events"
},
"sort_order": "desc",
"limit": 24
}
}
}
Output Schema
Window Results
{
"metadata": {
"windows": [
{
"window": "2026-04-01",
"metrics": {
"count": 3,
"avg_score": 0.85
}
},
{
"window": "2026-04-02",
"metrics": {
"count": 1,
"avg_score": 0.72
},
"drift": {
"absolute_change": -2,
"percent_change": -66.67,
"flagged": true
}
}
],
"num_windows": 2,
"num_documents_in": 4,
"num_documents_skipped": 0,
"window_granularity": "day"
}
}
Window Key Formats
| Window | Format | Example |
|---|---|---|
hour | YYYY-MM-DDTHH:00:00 | 2026-04-01T15:00:00 |
day | YYYY-MM-DD | 2026-04-01 |
week | YYYY-WNN | 2026-W14 |
month | YYYY-MM | 2026-04 |
quarter | YYYY-QN | 2026-Q2 |
year | YYYY | 2026 |
Performance
| Metric | Value |
|---|---|
| Latency | 5-50ms |
| Memory | O(windows x aggregations) |
| Cost | Free |
| Scalability | Efficient for large result sets |
Common Pipeline Patterns
Search + Temporal Analysis
[
{
"stage_name": "feature_search",
"stage_type": "filter",
"config": {
"stage_id": "feature_search",
"parameters": {
"searches": [
{
"feature_uri": "mixpeek://multimodal_extractor@v1/vertex_multimodal_embedding",
"query": "{{INPUT.query}}",
"top_k": 500
}
],
"final_top_k": 500
}
}
},
{
"stage_name": "temporal",
"stage_type": "reduce",
"config": {
"stage_id": "temporal",
"parameters": {
"time_field": "created_at",
"window": "day",
"aggregations": [
{"function": "count", "alias": "matches_per_day"},
{"function": "avg", "field": "score", "alias": "avg_relevance"}
],
"drift": {
"enabled": true,
"metric": "matches_per_day",
"threshold": 100.0
}
}
}
}
]
Brand Monitoring with Spike Detection
[
{
"stage_name": "feature_search",
"stage_type": "filter",
"config": {
"stage_id": "feature_search",
"parameters": {
"searches": [
{
"feature_uri": "mixpeek://multimodal_extractor@v1/vertex_multimodal_embedding",
"query": "{{INPUT.brand_name}}",
"top_k": 1000
}
],
"final_top_k": 1000
}
}
},
{
"stage_name": "llm_enrich",
"stage_type": "enrich",
"config": {
"stage_id": "llm_enrich",
"parameters": {
"provider": "google",
"model_name": "gemini-2.5-flash-lite",
"prompt": "Classify sentiment as positive, neutral, or negative: {{DOC.content}}",
"output_field": "sentiment"
}
}
},
{
"stage_name": "temporal",
"stage_type": "reduce",
"config": {
"stage_id": "temporal",
"parameters": {
"time_field": "published_at",
"window": "week",
"aggregations": [
{"function": "count", "alias": "mentions"},
{"function": "count_distinct", "field": "sentiment", "alias": "sentiment_spread"}
],
"drift": {
"enabled": true,
"metric": "mentions",
"threshold": 50.0
}
}
}
}
]
Error Handling
| Error | Behavior |
|---|---|
| Missing timestamp field | Document skipped |
| Unparseable timestamp | Document skipped |
| Non-numeric field for sum/avg | Document skipped for that aggregation |
| Empty results | 0 windows returned |
| Unknown aggregation | Returns null |