Stage Category: APPLY (Enriches documents)Transformation: N documents → N documents (with SQL data added)
When to Use
| Use Case | Description |
|---|---|
| Customer data enrichment | Join customer details from CRM database |
| Inventory lookups | Add real-time stock levels to product results |
| Pricing data | Enrich with current pricing from ERP |
| Cross-system joins | Combine vector search with relational data |
When NOT to Use
| Scenario | Recommended Alternative |
|---|---|
| Simple key-value lookups | document_enrich (collection join) |
| Read-heavy, cacheable data | Consider pre-indexing in Mixpeek |
| Real-time transactional queries | Direct database access |
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
connection_id | string | Required | Database connection identifier |
query | string | Required | SQL query with parameter placeholders |
parameters | object | {} | Mapping of placeholder names to document fields |
output_field | string | sql_result | Dot-path where query results are stored |
result_handling | string | first | How to handle multiple rows (first, all, error_if_empty, error_if_multiple) |
on_no_results | string | null | Behavior when no rows returned (skip, null, error) |
timeout | integer | 30 | Query timeout in seconds (1-300) |
on_error | string | skip | Error handling strategy (skip, remove, raise) |
Supported Databases
| Database | Connection Type | Notes |
|---|---|---|
| PostgreSQL | postgres | Full support |
| MySQL | mysql | Full support |
| SQL Server | mssql | Full support |
| SQLite | sqlite | Read-only |
Configuration Examples
{
"stage_name": "sql_lookup",
"stage_type": "apply",
"config": {
"stage_id": "sql_lookup",
"parameters": {
"connection_id": "crm_postgres",
"query": "SELECT name, email, tier FROM customers WHERE id = :customer_id",
"parameters": {
"customer_id": "{{DOC.metadata.customer_id}}"
},
"output_field": "customer_data"
}
}
}
{
"stage_name": "sql_lookup",
"stage_type": "apply",
"config": {
"stage_id": "sql_lookup",
"parameters": {
"connection_id": "inventory_db",
"query": "SELECT stock_level, warehouse_location, last_updated FROM inventory WHERE sku = :sku",
"parameters": {
"sku": "{{DOC.metadata.sku}}"
},
"output_field": "inventory",
"timeout": 5
}
}
}
{
"stage_name": "sql_lookup",
"stage_type": "apply",
"config": {
"stage_id": "sql_lookup",
"parameters": {
"connection_id": "orders_db",
"query": "SELECT order_id, status, total FROM orders WHERE customer_id = :cid ORDER BY created_at DESC LIMIT 5",
"parameters": {
"cid": "{{DOC.metadata.customer_id}}"
},
"output_field": "recent_orders",
"result_handling": "all"
}
}
}
{
"stage_name": "sql_lookup",
"stage_type": "apply",
"config": {
"stage_id": "sql_lookup",
"parameters": {
"connection_id": "product_db",
"query": "SELECT price, currency, discount_pct FROM pricing WHERE product_id = :pid AND region = :region",
"parameters": {
"pid": "{{DOC.document_id}}",
"region": "{{INPUT.user_region}}"
},
"output_field": "pricing"
}
}
}
Query Syntax
Parameter Placeholders
Use:name syntax for parameter placeholders:
SELECT * FROM users WHERE id = :user_id AND status = :status
Template Variables
| Variable | Description |
|---|---|
{{DOC.*}} | Any document field |
{{INPUT.*}} | Input parameters from retriever call |
{{CONTEXT.*}} | Pipeline context variables |
Output Schema
Single Row (default)
{
"document_id": "doc_123",
"content": "...",
"sql_result": {
"name": "John Doe",
"email": "john@example.com",
"tier": "premium"
}
}
Multiple Rows
{
"document_id": "doc_123",
"content": "...",
"recent_orders": [
{"order_id": "ord_1", "status": "delivered", "total": 99.99},
{"order_id": "ord_2", "status": "shipped", "total": 149.99}
]
}
No Results
{
"document_id": "doc_123",
"content": "...",
"sql_result": null
}
Security
SQL queries are parameterized to prevent injection attacks. Never concatenate user input directly into query strings.
| Security Feature | Description |
|---|---|
| Parameterized queries | All parameters are escaped |
| Connection isolation | Each connection uses dedicated credentials |
| Read-only option | Configure connections as read-only |
| Query timeout | Prevent long-running queries |
Performance
| Metric | Value |
|---|---|
| Latency | 10-100ms (depends on query complexity) |
| Connection pooling | Automatic |
| Parallel execution | Up to 10 concurrent queries |
| Timeout handling | Graceful with null result |
For high-volume lookups, ensure your database has appropriate indexes on the queried columns. Consider caching frequently accessed data.
Common Pipeline Patterns
Search + SQL Enrichment
[
{
"stage_name": "semantic_search",
"stage_type": "filter",
"config": {
"stage_id": "feature_search",
"parameters": {
"searches": [
{ "feature_uri": "mixpeek://text_extractor@v1/multilingual_e5_large_instruct_v1", "query": { "input_mode": "text", "value": "{{INPUT.query}}" }, "top_k": 20 }
],
"final_top_k": 20
}
}
},
{
"stage_name": "sql_lookup",
"stage_type": "apply",
"config": {
"stage_id": "sql_lookup",
"parameters": {
"connection_id": "product_db",
"query": "SELECT price, stock FROM products WHERE id = :id",
"parameters": {
"id": "{{DOC.metadata.product_id}}"
},
"output_field": "product_data"
}
}
},
{
"stage_name": "structured_filter",
"stage_type": "filter",
"config": {
"stage_id": "attribute_filter",
"parameters": {
"conditions": {
"field": "product_data.stock",
"operator": "gt",
"value": 0
}
}
}
}
]
Error Handling
| Error | Behavior |
|---|---|
| Query timeout | Handled per on_error (default: document passes through unchanged) |
| Connection failure | Handled per on_error (default: skip) |
| No rows returned | output_field set to null (default on_no_results) |
| Invalid SQL | Stage fails with error message |
Related
- Document Enrich - Collection-based joins
- API Call - REST API enrichment
- JSON Transform - Transform enriched data