How GreptimeDB's Pipeline ETL System Transforms Data During Ingestion

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.

Architecture and Entry Points

Ingestion begins at the HTTP event handler in 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:

---
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	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	User-defined field-to-column mapping via GreptimeTransformer
src/servers/src/pipeline.rs	run_pipeline entry point and result aggregation
src/servers/src/http/event.rs	HTTP handler building PipelineIngestRequest
src/pipeline/src/manager/pipeline_operator.rs	Pipeline storage and retrieval from the system table
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, while HTTP integration resides in 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. The HTTP handler resolves pipeline names to their YAML/JSON definitions before execution.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}