Common header schema v2

Every event the loader writes -- regardless of the transport it arrived on (a Kafka topic or a gRPC push) -- gets a standardised common header injected before insert. This is the base schema shared by all event tables in the DFE (Data Fusion Engine) pipeline. This page documents the full field set, the DDL, and the ORDER BY / PARTITION BY / codec / index decisions behind it.

Overview

Every event table shares a common header that provides:

1. Temporal ordering -- multiple timestamp fields for different use cases.
2. Multi-tenancy -- organisation ID for row-level security (RLS).
3. Deduplication -- time-ordered UUIDs for uniqueness.
4. Auditability -- original payload preservation.
5. Extensibility -- dynamic metadata via JSON fields.

Common header vs complete table schema

The common header is the base of every table's schema -- not the complete schema itself. The loader injects these system fields into every event.

• Landing table (dfe.main): schema IS just the common header. This is the catch-all for unrouted events.
• Other tables (e.g., dfe.auth, dfe.metrics): schema = common header plus data-specific columns, alongside the common header.

The loader creates no table, the landing table included. It reads the schema the deployment created (in DFE, the engine's schema phase) and holds records for a table that does not exist yet in its pending-schema buffer.

+----------------------------------+
|    landing table (dfe.main)      |   <- schema = common header ONLY
|  +----------------------------+  |
|  |      common header         |  |
|  |  (_timestamp, _org_id, ...) |  |
|  +----------------------------+  |
+----------------------------------+

+----------------------------------+
|  non-default table (dfe.auth)    |   <- schema = common header + data columns
|  +----------------------------+  |
|  |      common header         |  |
|  |  (_timestamp, _org_id, ...) |  |
|  +----------------------------+  |
|  |    data columns            |  |
|  |  (user_id, action, ip, ...) |  |
|  +----------------------------+  |
+----------------------------------+

The profile system (dfe-schemas common-header/*.yaml) controls which common header fields are injected. The profile does NOT define data-specific columns -- those come from the source data or are defined in the table's own DDL.

Common header DDL (default table)

This is the common header as the default table uses it. Non-default tables would include these columns alongside their own data-specific columns.

CREATE TABLE IF NOT EXISTS {db}.{table}
(
    -- Primary timestamps
    `_timestamp_load` DateTime64(3) DEFAULT now64(3) CODEC(Delta, ZSTD(1)),
    `_timestamp` DateTime64(3) CODEC(Delta, ZSTD(1)),
    `_timestamp_received` Nullable(DateTime64(3)) CODEC(Delta, ZSTD(1)),

    -- Identity
    `_uuid` UUID DEFAULT generateUUIDv7(),
    `_org_id` LowCardinality(String) CODEC(ZSTD(1)),

    -- Source identifier
    `_source` LowCardinality(String) CODEC(ZSTD(1)),

    -- Payload preservation
    `_raw` Nullable(String) CODEC(ZSTD(3)),
    `_json` Nullable(JSON(max_dynamic_paths = 2048)) CODEC(ZSTD(3)),

    -- Metadata
    `_tags` Nullable(JSON) CODEC(ZSTD(3)),

    -- Indexes
    INDEX idx_timestamp _timestamp TYPE minmax GRANULARITY 1,
    INDEX idx_raw_text _raw TYPE text(tokenizer = 'default') GRANULARITY 64
)
ENGINE = {engine}
ORDER BY (_org_id, _timestamp_load, _uuid)
PARTITION BY (toYYYYMM(_timestamp_load), _org_id)
SETTINGS index_granularity = 8192

Column reference

_timestamp_load

Property	Value
Type	DateTime64(3)
Nullable	No
Default	now64(3)
Codec	Delta, ZSTD(1)
Source	@generated: now64(3)

Purpose: When the data was loaded into ClickHouse.

Rationale:

• Primary query filter for operational dashboards ("show me events from the last hour").
• More reliable than event timestamp for time-windowed queries.
• ClickHouse generates via DEFAULT -- loader omits this field.
• All rows in a batch get the same load timestamp (acceptable trade-off).

Why first in ORDER BY after _org_id:

• Every dashboard query filters by recent time window.
• Combined with _org_id, provides excellent query locality.
• Enables efficient "tail -f" style queries.

_timestamp

Property	Value
Type	DateTime64(3)
Nullable	No
Default	None
Codec	Delta, ZSTD(1)
Source	@source: timestamp | now()

Purpose: When the event actually occurred (event time).

Rationale:

• Event time semantics for analytics and time-series analysis.
• May differ significantly from load time (buffered, delayed, replayed events).
• Fallback to now() ensures no NULL values in non-nullable column.
• minmax index enables efficient range scans.

Field resolution order:

1. timestamp field in source data.
2. @timestamp field (common in log formats).
3. _timestamp field.
4. Fallback to now() if all missing/invalid.

Validation:

• Must be within reasonable bounds (not before 1970, not far in future).
• Millisecond precision preserved.
• Auto-detection of epoch seconds/millis/micros/nanos.

Wire format:

An RFC3339 value -- 2026-09-07T05:53:53.385Z, which is what the receiver and both transforms emit -- is normalised to ClickHouse's own text form (2026-09-07 05:53:53.385, UTC) before the insert, and an offset is applied rather than truncated. Carried verbatim it stops the JSONEachRow DateTime64 reader at the zone suffix, which rejects the whole batch with code 27 and loses every row in it. The same normalisation applies to any DateTime/DateTime64 column the header pass fills, not only this one.

_timestamp_received

Property	Value
Type	Nullable(DateTime64(3))
Nullable	Yes
Default	None
Codec	Delta, ZSTD(1)
Source	@source: timestamp_received

Purpose: When the receiver/loader first received the event.

Rationale:

• Useful for latency analysis (time from generation to receipt).
• The loader reads the first hop's _timestamp_received stamp ahead of the column's directive, then the directive's fields, then the ingest time. No other key such as received_at is read unless the directive names it.
• Nullable because most sources do not provide this.
• Distinct from _timestamp_load which is insert time.

Use cases:

• Pipeline latency monitoring: _timestamp_load - _timestamp_received.
• Event age at receipt: _timestamp_received - _timestamp.
• SLA compliance tracking.

_uuid

Property	Value
Type	UUID
Nullable	No
Default	generateUUIDv7()
Codec	None (UUID is already compact)
Source	@generated: generateUUIDv7()

Purpose: Unique identifier for each event.

Rationale:

• UUIDv7 is time-ordered (millisecond precision + random suffix).
• Sortable within timestamp -- useful for deterministic ordering.
• Generated by ClickHouse DEFAULT -- no client-side generation needed.
• Last in ORDER BY to ensure uniqueness within (org_id, timestamp_load).

Why UUIDv7 over other options:

Option	Pros	Cons
UUIDv4	Widely supported	Not sortable, poor locality
UUIDv7	Time-ordered, sortable	Requires ClickHouse 24.8+
ULID	Time-ordered	Not native ClickHouse type
Snowflake	Compact, sortable	Requires coordination
xxhash64	Fast	Not guaranteed unique

ClickHouse 24.8+ functions:

-- Thread-monotonic (guarantees ordering within thread)
generateUUIDv7()
generateUUIDv7ThreadMonotonic()

-- Slightly faster, no ordering guarantee
generateUUIDv7NonMonotonic()

_org_id

Property	Value
Type	LowCardinality(String)
Nullable	No
Default	None
Codec	ZSTD(1)
Source	routing.org_id_field (default _org_id), never a column directive

Purpose: Organisation/tenant identifier for multi-tenancy and RLS.

Rationale:

• First in ORDER BY -- every query has an org_id filter (RLS).
• LowCardinality for dictionary encoding (orgs are highly repeated).
• Required field -- cannot be NULL. A record whose field is absent, null or empty is written under routing.default_org_id.
• Read from the field routing.org_id_field names and nothing else, so no column comment can point the tenant at a field a sender writes.

Row-Level Security (RLS):

-- Create row policy for tenant isolation
CREATE ROW POLICY org_isolation ON dfe.main
FOR SELECT
USING _org_id = currentUser()
TO ALL;

In a DFE deployment the engine's reconciler writes the row policies that scope each org's console users, so a deployment does not hand-write them.

Why LowCardinality:

• Typical deployment has <1000 unique orgs.
• Dictionary encoding reduces storage 10-100x for this column.
• Faster equality comparisons (dictionary index lookup).

_raw

Property	Value
Type	Nullable(String)
Nullable	Yes
Default	None
Codec	ZSTD(3)
Source	the record's own raw field: a receiver's raw capture, or an @renamed source

Purpose: Original raw data as received -- the data as it would appear in a tailed log file or a database row, BEFORE any RFC/format parsing.

Under full it is filled only when the source provides raw data, such as a syslog line the receiver captured. A JSON record is kept once, in _json, and _raw stays NULL.

NOT the same as _json: _raw is the original wire format (e.g., raw syslog RFC 3164/5424 line), while _json is the parsed/structured result.

Rationale:

• Captured BEFORE any transformation or parsing.
• Enables "grep-like" full-text searches across the original payload.
• Nullable -- can be disabled globally or per-table to save storage.
• Higher ZSTD level (3) for better compression of text.

Text search index (default, included in Common Header v2.1):

INDEX idx_raw_text _raw TYPE text(tokenizer = 'default') GRANULARITY 64

The text index (GA in ClickHouse 26.2+) provides deterministic full-text search with no false positives. Supported query functions: hasToken(), hasAnyTokens(), hasAllTokens(), LIKE, ILIKE. Insert overhead ~50%.

_raw without the index has no reason to exist. To avoid the overhead, disable _raw entirely via capture_mode -- do not keep the column without the index.

Capture modes control _raw and _json population:

capture_mode	_json	_raw	Use case
full (default)	Full payload (JSON type)	The record's own raw field, else NULL	Full observability
json_only	Full payload (JSON type)	NULL	Full observability without the _raw storage/index overhead
raw_only	NULL	Full Kafka payload (String)	CH CPU saving -- no JSON type overhead
extracted_only	NULL	NULL	Minimal -- only promoted schema fields

All four modes extract promoted fields to schema columns. Configurable at global, per-table, and DDL levels. See metadata.capture_mode in config.

_json

Property	Value
Type	Nullable(JSON(max_dynamic_paths = 2048))
Nullable	Yes
Default	None
Codec	ZSTD(3)
Source	not a directive -- see Capture modes above

Purpose: Complete Kafka message as native ClickHouse JSON type for structured path-based queries.

max_dynamic_paths: Default raised to 2048 (from ClickHouse default 1024) to cover multi-Beats deployments (ECS + Winlogbeat + Sysmon generates 500-1500 unique paths). Per-column DDL only -- cannot be changed after data is inserted. Hard max 10,000. Paths exceeding the limit are stored in shared data (slower queries, data not lost).

NOT the same as _raw: _json is the parsed/structured result stored as native columnar JSON. _raw is the original wire format before parsing.

Rationale:

• Captured BEFORE any transformation (preserves original structure).
• ClickHouse JSON type -- each JSON path stored as a native subcolumn.
• Enables ad-hoc queries on any field without schema changes.
• Nullable -- can be disabled to save storage.

ClickHouse JSON type benefits:

• Columnar storage within JSON (not stored as a string blob).
• Type inference per JSON path.
• Efficient queries: SELECT _json.user.id, _json.action FROM events.
• Supports nested objects and arrays.
• ClickHouse stores each path as a dense subcolumn with compression.

Requirements:

• ClickHouse 26.2+ for GA JSON type + text skip index (hard deck).

Configuration:

metadata:
  capture_mode: full  # full (default) | json_only | raw_only | extracted_only

capture_mode controls _json and _raw population -- see the Capture Modes section below. Legacy booleans capture_json / capture_raw are still accepted for backwards compatibility and mapped onto capture_mode.

_tags

Property	Value
Type	Nullable(JSON)
Nullable	Yes
Default	None
Codec	ZSTD(3)
Source	@source: first(tags/_tags/meta/metadata.tags)

Purpose: Metadata and collector/agent information.

Rationale:

• Config-driven source field list (first match wins).
• Stored as JSON for flexible querying.
• Contains collector metadata, enrichment tags, routing info.
• Nullable -- not all events have tags.

Field resolution order:

1. tags field.
2. _tags field.
3. meta field.
4. metadata.tags nested path.

Shape: the column is a ClickHouse JSON, which only accepts an object at the top level, and the loader shapes every value it writes into any JSON column to match -- whichever rule filled it (the tags_fields hoist or an @source/@renamed column comment) and on both insert formats, RowBinary and JSONEachRow. An object is stored as-is. An array -- ECS ships tags as an array of keywords -- is stored whole under a list key, so ["forwarded"] lands as {"list": ["forwarded"]} and queries read it as _tags.list. A scalar, such as a bare string, is stored under a value key. A string carrying JSON text is stored as what that text holds, and text that does not parse is stored as the string it is, under value.

Configuration:

metadata:
  tags_fields: [tags, _tags, meta, metadata.tags]
  tags_output: _tags
  drop_tags: false  # Remove source tags after extraction

Reading the wrapped list back

SELECT _tags.list.:`Array(Nullable(String))` AS tags
FROM dfe.main
WHERE arrayExists(x -> x = 'filebeat', _tags.list.:`Array(Nullable(String))`)

An object source passes through untouched, so _tags holding {"list": [...]} cannot be told apart from a source that sent that key itself.

Underscore prefix convention

All common header fields use an underscore prefix (_timestamp, _org_id, _uuid, etc.).

Rationale:

• Avoids collision with source data fields.
• Source data often contains timestamp, id, tags etc.
• Clear visual distinction between system and data fields.
• Consistent with ClickHouse system columns (_part, _partition_id).

Example collision prevention:

// Source event
{
  "timestamp": "2024-01-15T10:30:00Z",
  "id": "user-123",
  "tags": ["important", "urgent"]
}

// Stored as:
// _timestamp = 2024-01-15T10:30:00Z (from source timestamp)
// _uuid = 01234567-... (generated)
// _tags = {"list": ["important", "urgent"]} (from source tags, wrapped for the JSON column)
// + all original fields preserved in _json

ORDER BY design

ORDER BY (_org_id, _timestamp_load, _uuid)

Column order rationale

1. _org_id first:

   • Every query has an org_id filter (RLS enforcement).
   • Data physically grouped by organisation.
   • Partition pruning works with org_id in ORDER BY.

2. _timestamp_load second:

   • Primary time filter for operational queries.
   • Recent data queries are most common.
   • Good data locality for time-windowed dashboards.

3. _uuid last:

   • Ensures uniqueness within (org_id, timestamp_load).
   • Deterministic ordering for reproducible queries.
   • Enables efficient point lookups by UUID.

Query patterns optimised

-- Pattern 1: Recent events for an org (most common)
SELECT * FROM events
WHERE _org_id = 'acme' AND _timestamp_load > now() - INTERVAL 1 HOUR
-- Uses: Full ORDER BY prefix

-- Pattern 2: Event by UUID
SELECT * FROM events
WHERE _org_id = 'acme' AND _uuid = '...'
-- Uses: org_id prefix, UUID in last position

-- Pattern 3: Time range across orgs (admin dashboard)
SELECT _org_id, count() FROM events
WHERE _timestamp_load > now() - INTERVAL 1 DAY
GROUP BY _org_id
-- Uses: Scans all orgs but time-filtered

-- Pattern 4: Event time analysis
SELECT * FROM events
WHERE _org_id = 'acme' AND _timestamp BETWEEN '...' AND '...'
-- Uses: org_id prefix, minmax index on _timestamp

PARTITION BY design

PARTITION BY (toYYYYMM(_timestamp_load), _org_id)

Rationale

1. Monthly partitions by load time:

   • Natural TTL boundary (drop old months).
   • Reasonable partition count (~12-24 active).
   • Aligns with typical retention policies.

2. Includes _org_id:

   • Enables per-org partition pruning.
   • Efficient org-level data management.
   • Only acceptable for <100 orgs (partition cardinality limit).

Partition management

-- View partitions
SELECT partition, rows, bytes_on_disk
FROM system.parts
WHERE database = 'common' AND table = 'events'
ORDER BY partition;

-- Drop old data (by month)
ALTER TABLE common.events DROP PARTITION ('202401', 'acme');

-- Drop all data for an org
ALTER TABLE common.events DROP PARTITION ('*', 'acme');

-- TTL-based retention (alternative)
ALTER TABLE common.events MODIFY TTL _timestamp_load + INTERVAL 90 DAY;

Cardinality considerations

Orgs	Months	Partitions	Status
10	12	120	Excellent
50	12	600	Good
100	12	1,200	Acceptable
500	12	6,000	Too many -- use shared schema

For >100 orgs: use shared schema without org in partition:

PARTITION BY toYYYYMM(_timestamp_load)
-- RLS handles isolation, not partitioning

Codec selection

Delta + ZSTD(1) for timestamps

`_timestamp_load` DateTime64(3) CODEC(Delta, ZSTD(1))

Rationale:

• Delta encoding exploits the monotonic timestamp nature.
• Adjacent timestamps differ by small amounts.
• ZSTD(1) compresses the delta-encoded values.
• Level 1 balances speed and compression.

Compression ratio: typically 10-20x for timestamp columns.

ZSTD(3) for text/JSON

`_raw` Nullable(String) CODEC(ZSTD(3))
`_json` Nullable(JSON) CODEC(ZSTD(3))

Rationale:

• Higher level (3) for better compression of text.
• Text/JSON has high redundancy (field names, common values).
• Slightly slower compression but faster decompression.
• Worth the trade-off for storage-heavy columns.

Compression ratio: typically 5-15x for JSON, 3-8x for raw text.

No codec for UUID

`_uuid` UUID DEFAULT generateUUIDv7()

Rationale:

• UUID is already 16 bytes (compact binary format).
• UUIDv7 has some entropy that does not compress well.
• Codec overhead not worth the minimal compression gain.

Index strategy

Primary index (sparse)

The ORDER BY clause creates a sparse primary index:

ORDER BY (_org_id, _timestamp_load, _uuid)

• Index entry every index_granularity rows (default: 8192).
• ~1 index entry per 8KB-64KB of data.
• Extremely memory-efficient for large tables.

minmax index on _timestamp

INDEX idx_timestamp _timestamp TYPE minmax GRANULARITY 1

Purpose: efficient time range queries on event time (not load time).

How it works:

• Stores min/max _timestamp per granule.
• Query WHERE _timestamp BETWEEN x AND y skips granules outside range.
• GRANULARITY 1 = one min/max per index granularity (8192 rows).

When it helps:

• Queries filtering on event time, not load time.
• Analytics queries: "events that occurred in Q1".
• Audit queries: "what happened at timestamp X".

Text index on _raw (default)

-- Standard: text index (GA in ClickHouse 26.2+, required by hard deck)
INDEX idx_raw_text _raw TYPE text(tokenizer = 'default') GRANULARITY 64

The common header DDL carries this index on every _raw column. Under capture_mode: extracted_only _raw is NULL and the index has nothing to hold.

Trade-offs:

Index type	Pros	Cons
text (default)	GA in 26.2+, accurate, no false positives	~50% insert overhead on _raw
ngrambf_v1 (legacy)	Lower storage	False positives; no reason to prefer on 26.2+
None	Zero overhead	Full scan for text search -- use capture_mode = "extracted_only" instead

If you do not need full-text search on the raw payload, set capture_mode = "extracted_only" -- _raw is NULL and the text index is skipped entirely.

Engine selection

The schema supports three MergeTree variants:

SharedMergeTree (recommended)

ENGINE = SharedMergeTree()

Requirements: ClickHouse Cloud or 24.1+ with shared storage.

Benefits:

• Automatic replication without ZooKeeper.
• Adding a replica moves no data: every replica reads the same shared storage.
• No replica configuration needed.

ReplicatedMergeTree (clustered)

ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{db}/{table}', '{replica}')

Requirements: ZooKeeper or ClickHouse Keeper.

Benefits:

• Traditional replication with explicit control.
• Well-tested, stable.
• Works with any ClickHouse version.

MergeTree (single-node)

ENGINE = MergeTree()

Use case: development, testing, single-node deployments.

Benefits:

• Simplest configuration.
• No external dependencies.
• Fastest for local development.

The loader does not pick the engine: it writes to whatever table the deployment created.

Source field mapping

Using the DDL Expression Language (see ../clickhouse/ddl-directives.md):

Column	Expression	Description
_timestamp_load	@generated: now64(3)	DB generates on insert
_timestamp	@source: timestamp | now()	From source with fallback
_timestamp_received	@source: timestamp_received	Optional, from source
_uuid	@generated: generateUUIDv7()	DB generates on insert
_org_id	not a directive	routing.org_id_field (default _org_id), else routing.default_org_id
_raw	@source: first(raw/_raw)	The source's own raw data, never a JSON payload _json holds
_json	not a directive	Populated by capture_mode, not by a column expression
_tags	@source: first(tags/_tags/meta/metadata.tags)	First match wins

When the header pass promotes nothing

A payload the header pass cannot read -- not JSON, not a JSON object, or a valid object matching no column in the table -- once returned an empty field map and the row landed anyway, every column at its type default: no _source, no _tags, _timestamp at epoch zero. A row like that reads as data while carrying none, and cannot say where it came from.

The message is now rejected to the DLQ instead. Every one of those rejections logs an ERROR naming the table and the reason, at most once a minute, and counts dfe_loader_header_pass_skipped_total. The table name stays out of the metric's labels: it comes from a payload field with no allowlist, so labelling it would let untrusted input grow the label set without bound.

A table with a _json column keeps the whole payload, so a field no column holds is still stored. A table without one, such as dfe-hyperdx's own tables, keeps only its columns. Each top-level field that lands nowhere counts dfe_loader_fields_unstored_total, and a WARN names the table and the fields at most once a minute. DFE's own _ fields are not counted.

Configuration reference

Loader config

routing:
  org_id_field: _org_id             # Field to extract for _org_id
  default_org_id: default           # Tenant for a record with none
  org_routes: []                    # Empty = shared schema
  table_fields: [event_category]    # Field for table routing
  default_db: dfe
  default_table: main

metadata:
  capture_mode: full                # full (default) | json_only | raw_only | extracted_only
  tags_fields: [tags, _tags, meta, metadata.tags]
  tags_output: _tags
  drop_tags: false                  # Keep tags in _json after extraction

Implementation files

File	Purpose
src/clickhouse/schema.rs	Reads the deployed columns from system.columns on a TTL
src/column_meta/mod.rs	Parses the @directive annotations out of column comments
src/schema/mod.rs	Schema parsing, table tags, capability detection
src/transform/transformer.rs	Profile-driven field injection

The profile YAML itself lives in dfe-schemas under common-header/ and reaches the loader only as deployed ClickHouse columns -- see schemas.md.

Profiles

The common header is controlled by profiles -- named, versioned YAML definitions that specify which system fields to inject and how to populate them.

Profile	Fields	Use case
timeseries (default)	9	Full common header for event ingestion
minimal	5	High-volume structured data (no _raw,_tags, _source)
passthrough	4	Transparent bridge (no _timestamp injection)

Profiles define:

• Field set: which common header columns to inject.
• Field behaviour: source expression for each field (how it is populated).
• DDL structure: ORDER BY, PARTITION BY, indexes, for the table DDL the deployment creates.

Profiles do NOT define data-specific columns. Those come from the source data or are defined in the table's own DDL.

References

• ../clickhouse/ddl-directives.md -- field mapping expression language
• ClickHouse MergeTree
• ClickHouse JSON Type
• Row-Level Security