1 of 58

Building InfluxDB 3.0

(and other systems) without starting from “scratch”

with Apache DataFusion

Andrew Lamb | Staff Engineer, InfluxData

June 26, 2024. Microsoft Research / Gray Systems Lab

1

2 of 58

Andrew Lamb

Staff Engineer InfluxData

> 20 21 😱years in enterprise software development

Oracle: Database (2 years)

DataPower: XSLT compiler (2 years)

Vertica: DB / Query Optimizer (6 years)

Nutonian/DataRobot: ML Startups (7 years)

InfluxData: InfluxDB 3.0, Arrow, DataFusion (4 years)

MIT VI-2 2002, MEng 2003

i n f l u x d a t a . c o m

2

3 of 58

Outline

Database Implementation Trends
DataFusion: Introduction High Level Architecture
DataFusion in Action:

InfluxDB 3.0
Lance + LanceDB
Apache DataFusion Comet
SDF

Blog: Flight, DataFusion, Arrow, and Parquet: Using the FDAP Architecture to build InfluxDB 3.0

3

4 of 58

“One Size Fits All”

An Idea Whose Time Has Come and Gone

Stonebraker & Çetintemel (2005)

4

5 of 58

“One Size Fits All”: An Idea Whose Time Has Come and Gone*

Time

Adoption

Ease of Building

3.0

* Stonebraker & Çetintemel (2005)

5

6 of 58

It is hard (expensive) to build a new databases

Database company money raised

Snowflake: $2B
Databricks (Spark) $3.5B
MongoDB: $311M
SingleStore: $464.1M
CockroachLabs (CockroachDB): $633.1M
Pingcap (TiDB): $341.6M
Elastic: $162M
TimescaleDB: $181M
DuckDB $? / MotherDuck: $47.5M $100M

Source: https://www.crunchbase.com

I did this at Vertica too

6

7 of 58

“We can do it with the Apache Arrow Ecosystem”

https://www.influxdata.com/blog/apache-arrow-parquet-flight-and-their-ecosystem-are-a-game-changer-for-olap/

Evan (CEO)

Paul (CTO)

7

8 of 58

Implementation timeline for a new Database

8

Client

API

In memory

storage

In-Memory

filter + aggregation

Durability / persistence

Metadata Catalog +

Management

Query

Language

Parser

Optimized /

Compressed storage

Execution on

Compressed

Data

Joins!

Additional Client

Languages

Outer Joins

Subquery support

More advanced analytics

Cost based optimizer

Out of core algorithms

Storage Rearrangement

Heuristic Query Planner

Arithmetic expressions

Date / time Expressions

Concurrency

Control

Data Model /

Type System

Distributed query execution

Resource

Management

“Lets Build

a Database”

🤔

“Ok now this is pretty good”

😐

“Look mom!

I have a database!”

😃

Online recovery

Window functions

8

9 of 58

Bonus: Ecosystem Integration

In addition to lowering engineering costs…💰

Building on DataFusion (via Arrow / Flight / Parquet) gives instant ecosystem integration

(maybe this is really another example of lower costs 🤔)

Blog: Flight, DataFusion, Arrow, and Parquet: Using the FDAP Architecture to build InfluxDB 3.0

9

10 of 58

Ecosystem Integration

Where to connect?

Pandas

Drill

Impala

HBase

Kudu

Cassandra

Parquet

Spark

MyNewDB

10

11 of 58

Ecosystem Integration

MyNewDB

Flight

Use a common, efficient protocol.

Spark

Pandas

Drill

Impala

HBase

Kudu

Cassandra

Parquet

11

12 of 58

Ecosystem Integration

Object Store

“Lakehouse”

Data Lives in open formats on (very) cheap object storage

Specialized System 3

Specialized System 2

Specialized System 1

⇒ Native integration with Arrow/Flight/Parquet required for next gen systems

12

13 of 58

Intro to DataFusion

13

14 of 58

Apache 2.0 Licensed

Top Level Project of Apache Software Foundation (as of April 2024)

https://datafusion.apache.org/

14

15 of 58

What: DataFusion is LLVM for Databases

Analytic Application

Domain Specific Language

Specialized Database

Application Logic

Catalog

Analysis Engine

Multiple SQL Dialects

Data Flow Analysis

Custom Operators

File System Interface

…

Clang

Rust

Julia

Swift

C/C++ frontend

LLVM

…

Rustlang frontend

LLVM

Julia frontend

LLVM

Swift frontend

LLVM

15

16 of 58

Architecture

Design Goals:

Work “out of the box” (fast time to awesome)
Customize everything via APIs
Architecturally Boring 🥱 (“Industrial best practice”)

Results for Users

Quickly start with a basic, high-performance engine
Specialize to suit their needs and available engineering capacity
Easy to try out new ideas (operators, rewrites, etc)

16

17 of 58

DataFusion / Query Engine: Input / Output

17

Data Batches

SQL Query

SELECT status, COUNT(1)

FROM http_api_requests_total

WHERE path = '/api/v2/write'

GROUP BY status;

Data Batches

DataFrame

ctx.read_table("http")?

.filter(...)?

.aggregate(..)?;

Catalog information:

tables, schemas, etc

17

18 of 58

Architecture

SQL

Front Ends

DataFrame

LogicalPlan

ExecutionPlan

Plan Representations and Rewrites

Expression Eval

Optimizations / Transformations

HashAggregate

Sort

…

Execution Engine

Join

Catalog and

Data Sources

Parquet

CSV

…

Extension

Catalog / Table

Extension

Frontend

Extension

LogicalPlan Rewrite

Extension ExecutionPlan Rewrite

Extension

Stream

Extension Node

Streams

18

19 of 58

Case Study: InfluxDB 3.0

19

20 of 58

Time Series Databases: Crash Course

Specialized for storing data with times (obviously)

Key Properties:

Schema on Write: new columns can appear at any time, backfills
High volume, denormalized ingest

eg. host=myhost123.example.com repeated over and over

Rapid data value decay: newest data is super important, falls off drastically

Examples: InfluxDB (open), Azure Time Series Insights (closed), Facebook Gorilla (closed), Google Monarch (closed), Timescale (open), AWS Timestream (closed), Graphite Whisper / Grafana (open), …

20

21 of 58

InfluxDB 3.0 Requirements

Need: No series cardinality limits:

Tech: TSM (LSM Tree / KV Store) → Column Store

Need: ‘Infinite’ Retention

Tech: TSM on locally attached disks → object store (cheap!)

Need: Elastic Scalability

Tech: Shared Nothing (local disk)→ Disaggregated Storage (S3)

Need: Ecosystem Compatibility

Tech: InfluxQL / custom APIs → SQL + JDBC/ODBC

21

22 of 58

How InfluxDB 3.0 uses DataFusion

Rationale

Query and write are “just” moving data around
Reuse existing execution machinery (streaming, segregated worker pool, optimized sort / merge, etc)
Amplify investment in Open Source contributions

All queries and write path (parquet creation + compaction) use DataFusion + Arrow

22

23 of 58

InfluxDB 3.0 Architecture

Blog: InfluxDB 3.0: System Architecture

23

24 of 58

Query Processing in InfluxDB 3.0

Storage gRPC Frontend

SQL Frontend

Optimization

(storage pruning, pushdown, etc)

Physical Planning

Execution

gRPC output

Arrow Flight

Query Input

Client / Language Specific Frontends

Shared Planning, Execution

Phases, based on DataFusion

Output RecordBatches

Client Specific

Output formats

read_group(..)

SELECT …

FROM …

DataFusion

LogicalPlan

Arrow

Record Batches

Reorg Frontend

compact_plan(..)

ParquetWriter

SeriesFrame

...

FlightData

Write to Parquet files

DataFusion

LogicalPlan

DataFusion

ExecutionPlan

InfluxQL Frontend

SELECT … FROM … GROUP BY *

24

25 of 58

DataFusion extensions used in InfluxDB 3.0

SQL

Front Ends

DataFrame

LogicalPlan

ExecutionPlan

Plan Representations and Rewrites

Expression Eval

Optimizations / Transformations

HashAggregate

Sort

…

Execution Engine

Join

Catalog and

Data Sources

Parquet

CSV

…

Extension

Catalog / Table

Extension

Frontend

Extension

LogicalPlan Rewrite

Extension ExecutionPlan Rewrite

Extension

Stream

Extension Node

Streams

✅

25

26 of 58

Ingester

Persists data to object store as Parquet (sorted + deduplicated)

Object Store

Ingester

…

3. Periodically writes data buffer as sorted parquet files to object store

…

write path

1. Incoming line protocol

weather,location=us-east temperature=82,humidity=67 1465839830100400200�weather,location=us-midwest temperature=82,humidity=65 1465839830100400200�weather,location=us-west temperature=70,humidity=54 1465839830100400200�weather,location=us-east temperature=83,humidity=69 1465839830200400200�weather,location=us-midwest temperature=87,humidity=78 1465839830200400200�weather,location=us-west temperature=72,humidity=56 1465839830200400200�weather,location=us-east temperature=84,humidity=67 1465839830300400200�weather,location=us-midwest temperature=90,humidity=82 1465839830400400200�weather,location=us-west temperature=71,humidity=57 1465839830400400200

…

2. Parsed + Buffered in Arrow Compatible* format

* appendable until queried and then snapshotted

26

27 of 58

Ingester: Persist

DataFusion use: Sort and Deduplicate and write Parquet

Object Store

…

Ingester

Buffer

Scan

Sort

Dedupe

DataFusion Plan

1. Buffered data fed to plan

…

parquet-rs

writer

2. Plan output fed to parquet writer*

3. Parquet written to object store

* In the progress of migrating to use DataFusion parquet writer (to get parallelized encoding)

27

28 of 58

Ingester: Partition Pruning

DataFusion use: PruningPredicate for partition pruning

Buffer is partitioned (e.g. by user tag + time)

Stores min/max metadata

…

time: min/max

partn: min/max

…

time: min/max

partn: min/max

…

time: min/max

partn: min/max

Querier

SELECT … WHERE

time > now() - ‘1 minute’ AND

region = ‘us-east-1’

User SQL query

1. Querier parses / analyzes user SQL

time > 123456789

AND region = ‘us-east-1’

2. Predicate sent to ingester

4. Only matching partitions returned

Ingester

3. min/max evaluated using PruningPredicate

28

29 of 58

Compactor

DataFusion use: Merge / Deduplicate and write Parquet

Object Store

…

Compactor

* In the progress of migrating to use DataFusion parquet writer (to get parallelized encoding)

1. Multiple parquet files read from object store

…

parquet-rs

writer

3. Plan output fed to parquet writer*

DataFusion Plan

Read Parquet

Merge

Dedupe

2. DataFusion plan merges and deduplicates

29

30 of 58

Querier

DataFusion used throughout

Query Planner

Querier

SELECT … WHERE

time > now() - ‘1 minute’ AND

region = ‘us-east-1’

User SQL query

FlightSQL handler

(gRPC)

Metadata Cache

DataFusion

Catalog API

Remote Catalog Service

Ingester

…

Object Store

Data Cache

FlightSQL

1. FlightSQL query received

3. Execution Plan runs

2. Query planned using IOx catalog + ingester

DataFusion Plan

…

4. Results returned to user

30

31 of 58

Querier: Partition + File Pruning

DataFusion use: PruningPredicate for partition + file pruning

Query Planner

Querier

SELECT … WHERE

time > now() - ‘1 minute’ AND

region = ‘us-east-1’

User SQL query

Remote Catalog Service

Object Store

DataFusion

TableProvider API

Metadata Cache

Partition: 2024-06-01

002312.parquet

a773ef.parquet

599e92.parquet

Partition: 2024-06-02

0356af.parquet

54229a.parquet

B432f3.parquet

…

Partition: 2024-06-30

432790.parquet

4321aa.parquet

time > 123456789

AND region = ‘us-east-1’

2. Evaluate Pruning- Predicate with min/max

1. SQL planned, predicates pushed down

3. Only relevant partitions / files used for plan

31

32 of 58

Querier: Custom SQL Operator

cpu.user	time
89.3	2000-05-05T12:00:00Z
88.0	2000-05-05T12:05:01Z
85.0	2000-05-05T12:15:00Z
45.3	2000-05-05T12:34:00Z
...	...
78.9	2000-05-05T12:58:00Z
78.7	2000-05-05T12:59:00Z

Note: No data points for 12:20:00-12:29:00

cpu.user	time
88.6	2000-05-05T12:00:00Z
85.0	2000-05-05T12:10:01Z
???	2000-05-05T12:20:00Z
45.3	2000-05-05T12:30:00Z
55.0	2000-05-05T12:40:00Z
78.8	2000-05-05T12:50:00Z

SQL?

GROUP BY date_bin would have no row with this timestamp

32

33 of 58

Querier: Custom SQL Operator

SELECT

date_bin_gapfill(

interval '10 minute', time

) as minute,

avg(cpu.user)

from cpu

where

time between

timestamp '2000-05-05T12:00:00Z' and

timestamp '2000-05-05T12:59:00Z'

group by minute;

Implemented a date_bin_gapfill function modeled after time_bucket_gapfill

Implemented with Uses custom DataFusion operator and optimizer pass

33

34 of 58

Querier: Custom SQL Operator

SELECT

date_bin_gapfill(

interval '10 minute', time

) as minute,

avg(cpu.user)

from cpu

where

time between

timestamp '2000-05-05T12:00:00Z' and

timestamp '2000-05-05T12:59:00Z'

group by minute;

Filter

(time < '2000-05-05T12:00:00Z' and

time < '2000-05-05T12:59:00Z'

GroupBy

group: date_bin_gapfill(...)

agg: avg(cpu.user)

Scan

(projection + filter not shown)

Standard GroupBy can’t do gapfilling

34

35 of 58

Querier: Custom SQL Operator

Filter

(time < '2000-05-05T12:00:00Z' and

time < '2000-05-05T12:59:00Z'

GroupBy

group: minute

agg: date_bin_gapfill(...)

Scan

(projection + filter not shown)

Filter

(time < '2000-05-05T12:00:00Z' and

time < '2000-05-05T12:59:00Z'

GroupBy

group: date_bin as minute

agg: avg(...)

Scan

(projection + filter not shown)

GapFill

group: minute

agg: locf(avg(..))

1. OptimizerRule rewrites to use GapFill operator (extension)

Finalged expressions

35

36 of 58

Querier: Custom SQL Operator

ProjectionExec: expr=[minute, COUNT(cpu.user)] |

GapFillExec: group_expr=[date_bin_gapfill(..)],

aggr_expr=[COUNT(cpu.user)],

stride=10 minutes,

time_range=[957528000000000000).."957531540000000000"]

AggregateExec: mode=FinalPartitioned,

gby=[date_bin_gapfill(...)),cpu.time)],

aggr=[COUNT(cpu.user)]

CoalesceBatchesExec: target_batch_size=8192

RepartitionExec: partitioning=Hash([date_bin_gapfill(..., 4)

AggregateExec: mode=Partial,

gby=[date_bin_gapfill(...)),cpu.time)],

aggr=[COUNT(cpu.user)]

ParquetExec: file_groups=000000000000.parquet,predicates..)

Custom Stream (“operator)

36

37 of 58

Querier: Custom InfluxQL frontend

SELECT

MEAN(usage_idle),

MEAN(bytes_free)

FROM cpu, disk

GROUP BY TIME(10s)

FILL(linear)

SQL is kind of a pain to work with time series

InfluxDB has a DSL called “InfluxQL” that has SQL like syntax but specialized for time series

Not a cross join!

Gap filling + interpolation

37

38 of 58

Querier: Custom InfluxQL frontend

SELECT

MEAN(usage_idle),

MEAN(bytes_free)

FROM cpu, disk

GROUP BY TIME(10s)

FILL(linear)

InfluxQL Parser

(InfluxDB)

InfluxQL Planner

(InfluxDB)

1. InfluxQL Text

4. Optimized, Executed and Run like all other plans

3. Planned into a DataFusion LogicalPlan

Filter

(time < '2000-05-05T12:00:00Z' and

time < '2000-05-05T12:59:00Z'

GroupBy

group: date_bin as time

agg: avg(..), avg(..)

Scan

(projection + filter not shown)

GapFill

group: minute

agg: avg(..), avg(..)

LogicalPlan

2. Parsed into a custom AST

SELECT

AGG

Column

FROM

Table

…

usage_free

usage_idle

cpu

Table

disk

38

39 of 58

Querier: time series specific optimization

InfluxDB uses sortedness heavily, so often ends up with multi-column merges

Very wasteful for queries l

SELECT … FROM …

ORDER BY time DESC

LIMIT 10

SortPreservingMerge

expr: [tag1, tag2, time]

Input Stream

…

Each input stream ~ a partition, parquet files + data from ingester

More details in Blog: Making Most Recent Value Queries Hundreds of Times Faster

Not a cross join!

Have to read first batch from each stream before producing any output, even if limit

39

40 of 58

Querier: time series specific optimization

SortPreservingMerge

expr: [tag1, tag2, time]

ProgressiveEval

expr: [tag1, tag2, time]

Input Stream

…

* Really starts two in parallel to minimize IO stalls

2. Reads one stream at a time*

1. PhysicalOptimizerRule replaces SortPreservingMerge with ProgressiveEval if non overlapping key ranges

40

41 of 58

Case Study: Lance + LanceDB

Thanks to Weston Pace

41

42 of 58

Lance

Fast production-scale vector similarity, full-text & hybrid search and a SQL query interface (via DataFusion)

https://lancedb.github.io/lancedb

LanceDB: Open-source vector database for AI that's designed to store, manage, query and retrieve embeddings on large-scale multi-modal data.

Lance: open-source columnar data format designed for performant ML workloads and fast random access.

Both the database and the underlying data format are designed from the ground up to be easy-to-use, scalable and cost-effective.

Lance: File Format

LanceDB: Database

42

43 of 58

Lance Table Format

Protocol

Manifest

Implementation

Manifest - core spec describing what elements go in the table (lists of files, lists of secondary indices, statistics, etc.)
Protocol - rules describing how to update and change the manifest while respecting ACID
Implementation - the code implementing the protocol

44 of 58

DataFusion: Table Scanner

LanceDB doesn’t (yet) support full SQL but has a complicated planner that factors in which indices can be used in a query. This can generate

Over 1000 potential datafusion physical plans generated by Lance scanner

45 of 58

DataFusion: Beyond Scans

Custom DataFusion nodes implementing logic for secondary indices

Indexed_scan
Indexed_join
…

Upserts

indexed_scan -> full outer join

Training indices

full_scan -> out of core sort

let existing = session_ctx.read_lance(self.dataset.clone(), true)?;

let new_data = session_ctx.read_one_shot(source)?;

let join_cols = self

.params

.on // columns to join on

.iter()

.map(|c| c.as_str())

.collect::<Vec<_>>();

let joined = new_data.join(existing, JoinType::Full, &join_cols, &join_cols, None)?; // full join

Ok(joined.execute_stream().await?)

46 of 58

DataFusion: Even in the File Format!

Lance file format uses DataFusion to implement pushdown filters

Encoder uses DataFusion aggregators to calculate min/max (source)
Decoder uses DataFusion Expr simplification to calculate zone pruning (source)

47 of 58

Case Study: Apache DataFusion Comet

47

48 of 58

Apache DataFusion Comet

Runs Spark SQL queries using the native runtime (using Apache DataFusion execution)

⇒ faster and more resource efficient than JVM based runtimes.

Similar to

Databricks Photon Engine (proprietary)
Apache Gluten (incubating) - for Velox/Clickhouse

https://datafusion.apache.org/comet/

I just made up this icon!

48

49 of 58

Comet Goals

A native Parquet implementation, including both reader and writer
Full implementation of Spark operators, including Filter/Project/Aggregation/Join/Exchange etc.
Full implementation of Spark built-in expressions.
A UDF framework for users to migrate their existing UDF to native

49

50 of 58

Architecture

Integration Layer with Spark

DataFusion’s ExecutionPlan Streams

Use Spark Planner / Executor machinery

50

51 of 58

Architecture

51

52 of 58

Case Study: SDF

Thanks to Lukas Shulte

52

53 of 58

SDF

The Semantic Data Fabric

https://sdf.com

SDF is a multi-dialect SQL compiler, transformation framework,

and analytical database engine.

53

54 of 58

SDF

Assembly

Intelisense

Tests

Reports

Business Value

Catalog

The Key: Multi Dialect SQL -> Unified Logical Plan

BigQuery

Redshift

Metadata

Unified Logical Plan

SDF

Static Analysis

SDF Development Framework

Ingest

Deploy

Analyze

Guarantee

Snowflake

Trino/Presto

SDF uses complete ANTLR Grammars to define many SQL dialects - notably, proprietary ones like Snowflake.

All SQLs compile to a unified Intermediate Representation: the Datafusion Logical Plan.

This gives SDF Executable Semantics.

SDF’s transformation layer statically analyzes many logical plans at once for correctness and generates rich metadata.

54

55 of 58

SDF: Datafusion Enabled Transformation Layer

DBT Preprocesses macros and configurations.

Then issues a request to your cloud database.

The cloud database compiles the SQL and ultimately executes the query. Errors can ONLY be caught by the cloud database’s compiler.

SDF first reads your cloud database’s state.

Then statically analyzes all configuration (macros, metadata), queries (sql), and dataflows (classifiers) to provide compile time guarantees.

Only queries that have been validated both syntactically and semantically are run against the warehouse.

SDF Transformation

DBT Transformation

55

56 of 58

SDF: Integrated Database

Executable Semantics power SDF DB.

With a complete DF logical plan for any query in any environment , SDF can:

(a) Execute that query locally using Datafusion as the execution runtime

(b) Route that query to a remote cloud database

Enables execution of the same query against proprietary compute providers

56

57 of 58

The Result: Composable Execution

SDF provides intelligent federated execution between transformation layer and data cloud providers.

Composable Execution enables the same query to run against proprietary compute providers (like Snowflake, BigQuery, and Redshift) and SDF DB; with both query engines yielding the same result.

57

58 of 58

Thank you

Come and join us (it is lots of fun!)

100s of like-minded people hacking on Databases!

https://datafusion.apache.org/

58