Design a Time Series Database

1
# Prometheus exposition format
2
http_requests_total{method="GET",status="200"} 1234 1704067200000
3
http_requests_total{method="POST",status="500"} 5 1704067200000
4
cpu_usage{host="server-1",core="0"} 0.75 1704067200000
5

6
# InfluxDB line protocol
7
http_requests,method=GET,status=200 total=1234 1704067200000000000
8
cpu,host=server-1,core=0 usage=0.75 1704067200000000000

5 collapsed lines
1
message WriteRequest {
2
  repeated TimeSeries timeseries = 1;
3
  repeated MetricMetadata metadata = 2;
4
}
5

6
message TimeSeries {
7
  repeated Label labels = 1;  // [{name: "__name__", value: "http_requests_total"}, ...]
8
  repeated Sample samples = 2;
9
}
10

11
message Sample {
12
  double value = 1;
13
  int64 timestamp = 2;  // Unix ms
14
}

1
GET /api/v1/query_range?query=rate(http_requests_total{status="500"}[5m])&start=1704067200&end=1704153600&step=60

3 collapsed lines
1
{
2
  "status": "success",
3
  "data": {
4
    "resultType": "matrix",
5
    "result": [
6
      {
7
        "metric": {
8
          "__name__": "http_requests_total",
9
          "method": "GET",
10
          "status": "500",
11
          "instance": "server-1:9090"
12
        },
13
        "values": [
14
          [1704067200, "0.5"],
15
          [1704067260, "0.7"],
16
          [1704067320, "0.3"]
17
        ]
18
      }
19
    ]
20
  }
21
}

1
GET /api/v1/series?match[]=http_requests_total{status=~"5.."}

1
interface Sample {
2
  timestamp: number // Unix milliseconds
3
  value: number // IEEE 754 double (64-bit)
4
}
5

6
interface Series {
7
  labels: Map<string, string> // Metric name + tags
8
  samples: Sample[]
9
}
10

11
// Series identifier (sorted label pairs)
12
// http_requests_total{method="GET",status="200"}
13
// → __name__=http_requests_total,method=GET,status=200
14
// → Hash: fnv64a("__name__\xff http_requests_total\xff method\xff GET\xff ...")

1
block-<ulid>/
2
├── meta.json           # Block metadata
3
├── index               # Inverted index (labels → series)
4
├── chunks/
5
│   ├── 000001          # Compressed chunk file
6
│   ├── 000002
7
│   └── ...
8
└── tombstones          # Deleted series markers

3 collapsed lines
1
{
2
  "ulid": "01HQGJ5P1XXXXXXXXX",
3
  "minTime": 1704067200000,
4
  "maxTime": 1704074400000,
5
  "stats": {
6
    "numSamples": 50000000,
7
    "numSeries": 100000,
8
    "numChunks": 150000
9
  },
10
  "compaction": {
11
    "level": 1,
12
    "sources": ["01HQGJ..."]
13
  },
14
  "version": 1
15
}

1
Label → Series IDs
2
──────────────────
3
method=GET    → [1, 3, 5, 7, 9, ...]
4
method=POST   → [2, 4, 6, 8, ...]
5
status=200    → [1, 2, 3, 4, 5, ...]
6
status=500    → [6, 7, 8, ...]
7

8
Series ID → Chunk Refs
9
──────────────────────
10
1 → [(chunk_001, offset=0, len=4096), (chunk_002, offset=0, len=2048)]
11
2 → [(chunk_001, offset=4096, len=3072)]

1
http_requests{method="GET",status="500"}
2

3
1. Lookup method=GET → [1, 3, 5, 7, 9]
4
2. Lookup status=500 → [7, 8, 9, 10]
5
3. Intersect → [7, 9]
6
4. Fetch chunks for series 7, 9

5 collapsed lines
1
-- Series registry (the inverted index in memory/disk)
2
CREATE TABLE series (
3
    id BIGSERIAL PRIMARY KEY,
4
    labels_hash BIGINT UNIQUE NOT NULL,
5
    labels JSONB NOT NULL,
6
    created_at TIMESTAMPTZ DEFAULT NOW()
7
);
8

9
CREATE INDEX idx_series_labels ON series USING GIN (labels);
10

11
-- Samples (conceptual - stored in compressed chunks, not SQL)
12
CREATE TABLE samples (
13
    series_id BIGINT NOT NULL,
14
    timestamp TIMESTAMPTZ NOT NULL,
15
    value DOUBLE PRECISION NOT NULL,
16
    PRIMARY KEY (series_id, timestamp)
17
);
18

19
-- Block metadata
20
CREATE TABLE blocks (
21
    ulid TEXT PRIMARY KEY,
22
    min_time TIMESTAMPTZ NOT NULL,
23
    max_time TIMESTAMPTZ NOT NULL,
24
    num_samples BIGINT NOT NULL,
25
    num_series INTEGER NOT NULL,
26
    level INTEGER DEFAULT 1,
27
    created_at TIMESTAMPTZ DEFAULT NOW()
28
);
29

30
CREATE INDEX idx_blocks_time ON blocks(min_time, max_time);

8 collapsed lines
1
// Delta-of-delta encoding
2
// Timestamps: [1000, 1010, 1020, 1030, 1040]
3
// Deltas:     [  -,   10,   10,   10,   10]
4
// Delta-of-delta: [-, -, 0, 0, 0]
5

6
// Encoding rules:
7
// - If delta-of-delta = 0: write '0' (1 bit)
8
// - If fits in [-63, 64]: write '10' + 7 bits
9
// - If fits in [-255, 256]: write '110' + 9 bits
10
// - If fits in [-2047, 2048]: write '1110' + 12 bits
11
// - Otherwise: write '1111' + 32 bits
12

13
function encodeTimestamps(timestamps: number[]): BitStream {
14
  const stream = new BitStream()
15

16
  // First timestamp: stored as-is (64 bits)
17
  stream.writeBits(timestamps[0], 64)
18

19
  if (timestamps.length < 2) return stream
20

21
  // Second timestamp: delta from first
22
  let prevDelta = timestamps[1] - timestamps[0]
23
  stream.writeBits(prevDelta, 14) // Assuming max delta fits in 14 bits
24

25
  // Remaining: delta-of-delta
26
  for (let i = 2; i < timestamps.length; i++) {
27
    const delta = timestamps[i] - timestamps[i - 1]
28
    const deltaOfDelta = delta - prevDelta
29

30
    if (deltaOfDelta === 0) {
31
      stream.writeBit(0) // 1 bit for regular intervals
32
    } else if (deltaOfDelta >= -63 && deltaOfDelta <= 64) {
33
      stream.writeBits(0b10, 2)
34
      stream.writeBits(deltaOfDelta + 63, 7) // Offset to positive
35
    } else if (deltaOfDelta >= -255 && deltaOfDelta <= 256) {
36
      stream.writeBits(0b110, 3)
37
      stream.writeBits(deltaOfDelta + 255, 9)
38
    } else if (deltaOfDelta >= -2047 && deltaOfDelta <= 2048) {
39
      stream.writeBits(0b1110, 4)
11 collapsed lines
40
      stream.writeBits(deltaOfDelta + 2047, 12)
41
    } else {
42
      stream.writeBits(0b1111, 4)
43
      stream.writeBits(deltaOfDelta, 32)
44
    }
45

46
    prevDelta = delta
47
  }
48

49
  return stream
50
}
51

52
// Result: 96% of timestamps compress to 1 bit
53
// Average: 1.04 bits per timestamp (vs 64 bits uncompressed)

10 collapsed lines
1
// XOR encoding for IEEE 754 doubles
2
// Values: [0.75, 0.76, 0.76, 0.77]
3
// XOR with previous: [0.75, XOR(0.75,0.76), XOR(0.76,0.76), XOR(0.76,0.77)]
4

5
// The XOR of similar floats has many leading and trailing zeros
6
// We encode: (leading zeros count, significant bits count, significant bits)
7

8
interface XORState {
9
  prevValue: bigint // Previous value as 64-bit integer
10
  prevLeadingZeros: number
11
  prevTrailingZeros: number
12
}
13

14
function encodeValues(values: number[]): BitStream {
15
  const stream = new BitStream()
16

17
  // First value: stored as-is (64 bits)
18
  const firstBits = floatToBits(values[0])
19
  stream.writeBits(firstBits, 64)
20

21
  let state: XORState = {
22
    prevValue: firstBits,
23
    prevLeadingZeros: 0,
24
    prevTrailingZeros: 0,
25
  }
26

27
  for (let i = 1; i < values.length; i++) {
28
    const currentBits = floatToBits(values[i])
29
    const xor = currentBits ^ state.prevValue
30

31
    if (xor === 0n) {
32
      // Identical value: write '0' (1 bit)
33
      stream.writeBit(0)
34
    } else {
35
      stream.writeBit(1) // Values differ
36

37
      const leadingZeros = countLeadingZeros(xor)
38
      const trailingZeros = countTrailingZeros(xor)
39
      const significantBits = 64 - leadingZeros - trailingZeros
40

41
      if (leadingZeros >= state.prevLeadingZeros && trailingZeros >= state.prevTrailingZeros) {
42
        // Fits in previous window: write '0' + significant bits
43
        stream.writeBit(0)
44
        const shift = 64 - state.prevLeadingZeros - significantBits
45
        stream.writeBits(xor >> BigInt(shift), significantBits)
46
      } else {
47
        // New window: write '1' + leading zeros + length + bits
48
        stream.writeBit(1)
49
        stream.writeBits(leadingZeros, 5)
11 collapsed lines
50
        stream.writeBits(significantBits - 1, 6) // -1 because min is 1
51
        stream.writeBits(xor >> BigInt(trailingZeros), significantBits)
52

53
        state.prevLeadingZeros = leadingZeros
54
        state.prevTrailingZeros = trailingZeros
55
      }
56
    }
57

58
    state.prevValue = currentBits
59
  }
60

61
  return stream
62
}
63

64
// Result:
65
// - 51% of values compress to 1 bit (identical to previous)
66
// - 30% compress to ~27 bits (control '10' path)
67
// - 19% compress to ~37 bits (control '11' path)
68
// Average: 1.33 bytes per value (vs 8 bytes uncompressed)

10 collapsed lines
1
// WAL segment structure
2
// Segments are 128MB each, kept for minimum 3 files
3

4
interface WALSegment {
5
  id: number
6
  path: string
7
  size: number
8
  firstTimestamp: number
9
  lastTimestamp: number
10
}
11

12
interface WALRecord {
13
  type: "series" | "samples" | "tombstone"
14
  data: Uint8Array
15
}
16

17
class WriteAheadLog {
18
  private currentSegment: WALSegment
19
  private segments: WALSegment[] = []
20
  private readonly maxSegmentSize = 128 * 1024 * 1024 // 128MB
21
  private readonly minSegmentsKept = 3
22

23
  async append(records: WALRecord[]): Promise<void> {
24
    // Batch encode records
25
    const encoded = this.encodeRecords(records)
26

27
    // Check if current segment is full
28
    if (this.currentSegment.size + encoded.length > this.maxSegmentSize) {
29
      await this.rotateSegment()
30
    }
31

32
    // Sync write to disk (critical for durability)
33
    await this.currentSegment.write(encoded)
34
    await this.currentSegment.sync() // fsync
35
  }
36

37
  async truncate(checkpointSegmentId: number): Promise<void> {
38
    // Remove segments before checkpoint, keeping minimum
39
    const toRemove = this.segments.filter(
40
      (s) => s.id < checkpointSegmentId && this.segments.length - 1 >= this.minSegmentsKept,
41
    )
42

43
    for (const segment of toRemove) {
44
      await fs.unlink(segment.path)
11 collapsed lines
45
      this.segments = this.segments.filter((s) => s.id !== segment.id)
46
    }
47
  }
48

49
  async replay(): Promise<WALRecord[]> {
50
    // On startup, replay all records from WAL
51
    const records: WALRecord[] = []
52
    for (const segment of this.segments) {
53
      const segmentRecords = await this.readSegment(segment)
54
      records.push(...segmentRecords)
55
    }
56
    return records
57
  }
58
}

10 collapsed lines
1
// Inverted index for label lookups
2
// Uses posting lists (sorted arrays of series IDs)
3

4
type SeriesID = number
5
type PostingList = SeriesID[] // Sorted, deduplicated
6

7
class InMemoryIndex {
8
  // Label name → label value → posting list
9
  private postings: Map<string, Map<string, PostingList>> = new Map()
10

11
  // Series ID → labels
12
  private seriesLabels: Map<SeriesID, Map<string, string>> = new Map()
13

14
  // Labels hash → Series ID (for deduplication)
15
  private hashToSeries: Map<bigint, SeriesID> = new Map()
16

17
  private nextSeriesID: SeriesID = 1
18

19
  getOrCreateSeries(labels: Map<string, string>): SeriesID {
20
    const hash = this.hashLabels(labels)
21

22
    const existing = this.hashToSeries.get(hash)
23
    if (existing !== undefined) {
24
      return existing
25
    }
26

27
    // Create new series
28
    const id = this.nextSeriesID++
29
    this.hashToSeries.set(hash, id)
30
    this.seriesLabels.set(id, labels)
31

32
    // Update posting lists
33
    for (const [name, value] of labels) {
34
      if (!this.postings.has(name)) {
35
        this.postings.set(name, new Map())
36
      }
37
      const valueMap = this.postings.get(name)!
38
      if (!valueMap.has(value)) {
39
        valueMap.set(value, [])
40
      }
41
      valueMap.get(value)!.push(id)
42
    }
43

44
    return id
45
  }
46

47
  // Query: find series matching label matchers
48
  query(matchers: LabelMatcher[]): SeriesID[] {
49
    if (matchers.length === 0) {
50
      return Array.from(this.seriesLabels.keys())
51
    }
52

53
    // Start with smallest posting list (most selective)
54
    const sorted = [...matchers].sort((a, b) => this.getPostingListSize(a) - this.getPostingListSize(b))
11 collapsed lines
55

56
    let result = this.getPostingList(sorted[0])
57

58
    // Intersect with remaining matchers
59
    for (let i = 1; i < sorted.length; i++) {
60
      const other = this.getPostingList(sorted[i])
61
      result = this.intersect(result, other)
62

63
      if (result.length === 0) break // Early exit
64
    }
65

66
    return result
67
  }
68

69
  private intersect(a: PostingList, b: PostingList): PostingList {
70
    // Sorted merge intersection: O(n + m)
71
    const result: SeriesID[] = []
72
    let i = 0,
73
      j = 0
74

75
    while (i < a.length && j < b.length) {
76
      if (a[i] === b[j]) {
77
        result.push(a[i])
78
        i++
79
        j++
80
      } else if (a[i] < b[j]) {
81
        i++
82
      } else {
83
        j++
84
      }
85
    }
86

87
    return result
88
  }
89
}

10 collapsed lines
1
// Compaction strategy
2
// - Level 0: 2-hour blocks (from memtable flush)
3
// - Level 1: 8-hour blocks (4 level-0 merged)
4
// - Level 2: 24-hour blocks
5
// - Max block size: 31 days OR 10% of retention
6

7
interface CompactionPlan {
8
  sources: Block[]
9
  targetLevel: number
10
  estimatedSize: number
11
}
12

13
class Compactor {
14
  private readonly maxBlockDuration = 31 * 24 * 60 * 60 * 1000 // 31 days
15

16
  async planCompaction(blocks: Block[]): Promise<CompactionPlan[]> {
17
    const plans: CompactionPlan[] = []
18

19
    // Group blocks by level
20
    const byLevel = this.groupByLevel(blocks)
21

22
    for (const [level, levelBlocks] of byLevel) {
23
      // Find adjacent blocks that can be merged
24
      const sorted = levelBlocks.sort((a, b) => a.minTime - b.minTime)
25

26
      for (let i = 0; i < sorted.length - 1; i++) {
27
        const candidates = [sorted[i]]
28
        let j = i + 1
29

30
        while (j < sorted.length && sorted[j].minTime === candidates[candidates.length - 1].maxTime) {
31
          candidates.push(sorted[j])
32
          j++
33

34
          // Check if merged duration exceeds max
35
          const duration = candidates[j - 1].maxTime - candidates[0].minTime
36
          if (duration > this.maxBlockDuration) break
37

38
          // Typical: merge 4 blocks at a time
39
          if (candidates.length >= 4) break
40
        }
41

42
        if (candidates.length >= 2) {
43
          plans.push({
44
            sources: candidates,
45
            targetLevel: level + 1,
46
            estimatedSize: candidates.reduce((sum, b) => sum + b.size, 0) * 0.9,
47
          })
48
          i = j - 1 // Skip merged blocks
49
        }
11 collapsed lines
50
      }
51
    }
52

53
    return plans
54
  }
55

56
  async compact(plan: CompactionPlan): Promise<Block> {
57
    // 1. Open iterators for all source blocks
58
    const iterators = plan.sources.map((b) => b.createIterator())
59

60
    // 2. Merge-sort by (seriesID, timestamp)
61
    const mergedSamples = this.mergeSortIterators(iterators)
62

63
    // 3. Write to new block with fresh compression
64
    const newBlock = await this.writeBlock(mergedSamples, plan.targetLevel)
65

66
    // 4. Update block references atomically
67
    await this.swapBlocks(plan.sources, newBlock)
68

69
    // 5. Delete source blocks
70
    for (const source of plan.sources) {
71
      await source.delete()
72
    }
73

74
    return newBlock
75
  }
76
}

10 collapsed lines
1
// Downsampling tiers
2
// - Raw: 10-second resolution, 15 days
3
// - 1-minute: min/max/sum/count/avg, 90 days
4
// - 1-hour: min/max/sum/count/avg, 2 years
5

6
interface DownsampleConfig {
7
  sourceResolution: string // "raw", "1m", "1h"
8
  targetResolution: string
9
  aggregations: string[] // ["min", "max", "sum", "count"]
10
  retention: number // milliseconds
11
}
12

13
interface DownsampledSample {
14
  timestamp: number // Bucket start time
15
  min: number
16
  max: number
17
  sum: number
18
  count: number
19
  // Derived: avg = sum / count
20
}
21

22
class Downsampler {
23
  async downsample(seriesId: number, sourceBlock: Block, config: DownsampleConfig): Promise<DownsampledSample[]> {
24
    const bucketMs = this.parseDuration(config.targetResolution)
25
    const buckets = new Map<number, DownsampledSample>()
26

27
    const iterator = sourceBlock.createIterator(seriesId)
28

29
    for await (const sample of iterator) {
30
      const bucketStart = Math.floor(sample.timestamp / bucketMs) * bucketMs
31

32
      if (!buckets.has(bucketStart)) {
33
        buckets.set(bucketStart, {
34
          timestamp: bucketStart,
35
          min: sample.value,
36
          max: sample.value,
37
          sum: sample.value,
38
          count: 1,
39
        })
40
      } else {
41
        const bucket = buckets.get(bucketStart)!
42
        bucket.min = Math.min(bucket.min, sample.value)
43
        bucket.max = Math.max(bucket.max, sample.value)
44
        bucket.sum += sample.value
11 collapsed lines
45
        bucket.count++
46
      }
47
    }
48

49
    return Array.from(buckets.values())
50
  }
51
}
52

53
// Query routing: select appropriate resolution based on query range
54
function selectResolution(queryRange: number): string {
55
  if (queryRange <= 24 * 60 * 60 * 1000) {
56
    // ≤ 1 day
57
    return "raw"
58
  } else if (queryRange <= 30 * 24 * 60 * 60 * 1000) {
59
    // ≤ 30 days
60
    return "1m"
61
  } else {
62
    return "1h"
63
  }
64
}

8 collapsed lines
1
// Cardinality tracking and enforcement
2

3
interface CardinalityLimits {
4
  maxSeriesPerMetric: number // e.g., 100,000
5
  maxTotalSeries: number // e.g., 10,000,000
6
  maxLabelValueCardinality: number // e.g., 10,000 per label name
7
}
8

9
class CardinalityTracker {
10
  private seriesPerMetric: Map<string, Set<number>> = new Map()
11
  private seriesPerLabel: Map<string, Map<string, number>> = new Map()
12
  private totalSeries: number = 0
13

14
  checkLimits(labels: Map<string, string>, limits: CardinalityLimits): CardinalityError | null {
15
    const metricName = labels.get("__name__")!
16

17
    // Check per-metric cardinality
18
    const metricSeries = this.seriesPerMetric.get(metricName)?.size ?? 0
19
    if (metricSeries >= limits.maxSeriesPerMetric) {
20
      return {
21
        type: "metric_cardinality_exceeded",
22
        metric: metricName,
23
        current: metricSeries,
24
        limit: limits.maxSeriesPerMetric,
25
      }
26
    }
27

28
    // Check per-label cardinality (detect high-cardinality labels)
29
    for (const [name, value] of labels) {
30
      const labelValues = this.seriesPerLabel.get(name)
31
      if (labelValues && labelValues.size >= limits.maxLabelValueCardinality) {
32
        return {
33
          type: "label_cardinality_exceeded",
34
          label: name,
35
          current: labelValues.size,
36
          limit: limits.maxLabelValueCardinality,
37
        }
38
      }
39
    }
11 collapsed lines
40

41
    // Check total cardinality
42
    if (this.totalSeries >= limits.maxTotalSeries) {
43
      return {
44
        type: "total_cardinality_exceeded",
45
        current: this.totalSeries,
46
        limit: limits.maxTotalSeries,
47
      }
48
    }
49

50
    return null
51
  }
52

53
  // Identify high-cardinality labels for alerting
54
  getHighCardinalityLabels(threshold: number): { label: string; cardinality: number }[] {
55
    const results: { label: string; cardinality: number }[] = []
56

57
    for (const [label, values] of this.seriesPerLabel) {
58
      if (values.size > threshold) {
59
        results.push({ label, cardinality: values.size })
60
      }
61
    }
62

63
    return results.sort((a, b) => b.cardinality - a.cardinality)
64
  }
65
}
66

67
// Common high-cardinality antipatterns to detect:
68
// - request_id, trace_id, user_id as labels (use exemplars instead)
69
// - IP addresses, URLs with query strings
70
// - Timestamps or UUIDs in labels

5 collapsed lines
1
// Dashboard query batching
2
interface DashboardQuery {
3
  panelId: string
4
  expr: string
5
  range: { start: number; end: number }
6
  step: number
7
}
8

9
async function fetchDashboard(queries: DashboardQuery[]): Promise<Map<string, QueryResult>> {
10
  // Group queries by time range (likely the same for most panels)
11
  const byRange = groupBy(queries, (q) => `${q.range.start}-${q.range.end}`)
12

13
  const results = new Map<string, QueryResult>()
14

15
  // Fetch each range group in parallel
16
  await Promise.all(
17
    Object.entries(byRange).map(async ([rangeKey, rangeQueries]) => {
18
      // Batch multiple expressions if backend supports it
19
      const batchResult = await fetchBatch(rangeQueries)
20

21
      for (const [panelId, result] of batchResult) {
22
        results.set(panelId, result)
23
      }
24
    }),
25
  )
26

27
  return results
28
}