# How GreptimeDB's Pipeline ETL System Transforms Data During Ingestion

> GreptimeDB's pipeline ETL system transforms data during ingestion using VRL processors and dispatcher rules. Learn how to map fields to typed columns with schemas or inference.

- Repository: [Greptime/greptimedb](https://github.com/greptimeteam/greptimedb)
- Tags: deep-dive
- Published: 2026-03-02

---

**GreptimeDB's pipeline ETL system transforms incoming log data by executing a YAML-defined sequence of VRL-based processors, applying optional dispatcher rules for conditional routing, and mapping the processed fields to typed column values through either a user-defined schema or automatic inference.**

The pipeline ETL system in the `greptimeteam/greptimedb` repository provides a server-side transformation layer that intercepts raw log ingest requests and converts them into structured storage rows. When a client sends a request to `POST /v1/ingest?pipeline_name=demo`, the server constructs a `PipelineIngestRequest` containing the target table name and raw log records as **VRL** values (`vrl::core::Value`), then executes a multi-stage transformation pipeline defined in [`src/pipeline/src/etl.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/pipeline/src/etl.rs).

## Architecture and Entry Points

Ingestion begins at the HTTP event handler in [`src/servers/src/http/event.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/servers/src/http/event.rs), which builds a `PipelineIngestRequest` from the incoming JSON payload. The server-side entry point `run_pipeline` (located at `src/servers/src/pipeline.rs#L65-L77`) determines whether to use the built-in identity pipeline or a custom user-defined pipeline, then forwards the request to `Pipeline::exec_mut` for execution.

## Loading Pipeline Definitions

The system parses pipeline configurations using `Pipeline::parse`, which reads YAML or JSON descriptions to create a `Pipeline` struct. According to the source code at `src/pipeline/src/etl.rs#L60-L89`, this struct contains three core components:

- **Processors**: A list of transformation steps such as `urlencoding`, `epoch`, and `regex` that mutate individual fields
- **Dispatcher**: Optional routing rules that can redirect rows to alternate tables based on field values (`src/pipeline/src/etl.rs#L19-L24`)
- **Transformer**: The final mapping engine, which can be either `GreptimeTransformer` for user-defined schemas or `AutoTransform` for automatic inference (`src/pipeline/src/etl.rs#L86-L113`)

## The Execution Pipeline

The ETL engine processes data through three distinct stages:

### 1. Processor Stage

The `Pipeline::exec_mut` method iterates sequentially over every configured processor in the order defined in the YAML. Each processor mutates the current `VrlValue` in place. As implemented at `src/pipeline/src/etl.rs#L11-L17`, if any processor returns `null`, the entire log line is filtered out and dropped from the batch.

### 2. Dispatcher Stage

After processing, the engine evaluates dispatcher rules. If a rule matches the current row's field values, the engine immediately returns a `DispatchedTo` result, bypassing further transformation and routing the row to the specified target table (`src/pipeline/src/etl.rs#L19-L22`). This allows conditional routing without additional processing overhead.

### 3. Transformation Stage

Before transformation, the engine normalizes input to an array format—single objects are wrapped in one-element arrays (`src/pipeline/src/etl.rs#L24-L28`). The system then applies one of two transformation modes:

**GreptimeTransformer** calls `transform_array_elements_by_ctx` to process each element. For every array item, it invokes `GreptimeTransformer::transform_mut` to produce a map of column names to `ValueData` types (`src/pipeline/src/etl.rs#L12-L14`). It then extracts a `ContextOpt` containing per-row options like custom table suffixes (`src/pipeline/src/etl.rs#L16-L20`), resolves the final table name by combining the original table with any suffix (`src/pipeline/src/etl.rs#L21-L24`), and emits a `Row { values }` grouped by the `(ContextOpt, table_name)` tuple (`src/pipeline/src/etl.rs#L24-L27`).

**AutoTransform** creates a temporary `PipelineContext` with an automatic *epoch* time index, then uses the helper `values_to_rows` to convert the raw VRL object plus the inferred timestamp into storage-ready rows (`src/pipeline/src/etl.rs#L42-L55`).

## Result Aggregation and Insertion

All generated rows are collected in a `ContextReq` structure, which drives the actual insert operation (`src/servers/src/pipeline.rs#L84-L99`). The pipeline stage itself does **not** persist data; it only produces correctly shaped `RowInsertRequest` objects for the downstream storage engine (Mito/Apache Arrow).

## Pipeline Configuration Example

The following YAML demonstrates a typical log processing pipeline:

```yaml
---
description: Demo log pipeline
processors:
  - urlencoding:
      fields: [breadcrumbs, UA, referer, queryStr]
      method: decode
      ignore_missing: true
  - epoch:
      field: reqTimeSec
      resolution: second
      ignore_missing: true
  - regex:
      field: breadcrumbs
      patterns:
        - "(?<parent>\\[[^\\[]*c=c[^\\]]*\\])"
        - "(?<edge>\\[[^\\[]*c=g[^\\]]*\\])"
transform:
  - fields: [breadcrumbs, referer, queryStr, customField]
    type: string
  - fields: [version, cacheStatus, lastByte]
    type: uint8

```

## Key Implementation Files

| File | Role |
|------|------|
| [`src/pipeline/src/etl.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/pipeline/src/etl.rs) | Core pipeline executor (`exec_mut`), dispatcher hook, and transformation orchestration |
| `src/pipeline/src/etl/processor/*.rs` | Individual processor implementations (regex, urlencoding, epoch) |
| [`src/pipeline/src/etl/transformer/greptime.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/pipeline/src/etl/transformer/greptime.rs) | User-defined field-to-column mapping via `GreptimeTransformer` |
| [`src/servers/src/pipeline.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/servers/src/pipeline.rs) | `run_pipeline` entry point and result aggregation |
| [`src/servers/src/http/event.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/servers/src/http/event.rs) | HTTP handler building `PipelineIngestRequest` |
| [`src/pipeline/src/manager/pipeline_operator.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/pipeline/src/manager/pipeline_operator.rs) | Pipeline storage and retrieval from the system table |
| [`src/pipeline/src/etl/dispatcher.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/pipeline/src/etl/dispatcher.rs) | Rule engine for conditional table routing |

## Summary

- The **pipeline ETL system** executes a three-layer transformation: field-level processors, optional dispatcher routing, and schema mapping.
- **Processors** mutate VRL values sequentially in `Pipeline::exec_mut`, with `null` returns filtering out rows.
- The **dispatcher** can short-circuit processing and route rows to alternate tables via `DispatchedTo` results.
- **GreptimeTransformer** provides custom schema mapping with per-row `ContextOpt` support, while **AutoTransform** infers timestamps automatically.
- All transformation logic lives in [`src/pipeline/src/etl.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/pipeline/src/etl.rs), while HTTP integration resides in [`src/servers/src/pipeline.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/servers/src/pipeline.rs).

## Frequently Asked Questions

### What is the difference between GreptimeTransformer and AutoTransform?

**GreptimeTransformer** requires explicit field mappings in the YAML configuration, allowing precise control over column names, data types, and table suffixes via `ContextOpt`. **AutoTransform** automatically creates a timestamp column using epoch time and leaves other fields as-is, requiring no schema definition for rapid prototyping.

### How does the dispatcher route rows to different tables?

The dispatcher evaluates rules after the processor stage. When a rule matches field values in the current row, it returns a `DispatchedTo` result immediately (`src/pipeline/src/etl.rs#L19-L22`), bypassing the standard transformer and sending the row directly to the target table specified in the rule.

### What happens when a processor returns null?

If any processor in the chain returns `null` for a log line, `Pipeline::exec_mut` filters out that entire row (`src/pipeline/src/etl.rs#L11-L17`). This provides a mechanism for dropping malformed or irrelevant data during ingestion.

### Where are pipeline definitions stored in GreptimeDB?

Pipeline definitions are stored and retrieved from an internal system table via the `pipeline_operator` module in [`src/pipeline/src/manager/pipeline_operator.rs`](https://github.com/greptimeteam/greptimedb/blob/main/src/pipeline/src/manager/pipeline_operator.rs). The HTTP handler resolves pipeline names to their YAML/JSON definitions before execution.