Using Sorting in Database Systems
Andrew Lamb, InfluxData
Robust Database Queries and Stream Processing
Lightning Talk, Dagstuhl Seminar 26311, 2026-07-28
Outline
How similar are sorting and streaming in data systems?
Prior Work
“Goetz has probably forgotten more of this topic than I will ever know”
Lamb Conjecture: Sorting in Analytic Systems
“Taking advantage of sortedness is a key analytic system feature”
High Data Volume ⇒ high cost secondary indexes (e.g. sorted B-Trees)
⇒ Sorting primary storage to organize data (including Z ordering, etc)
⇒ but not guaranteed as sorting is so expensive (and have to pick one order)
Fully Sorted Data
Full Sort: Point / Range Lookups
SELECT avg(temp)
FROM weather
WHERE time BETWEEN '12:00:00' AND '12:05:00'
12:01:00
12:02:00
12:03:00
12:06:00
88.3
89.3
88.0
91.1
12:07:00
52.4
12:08:00
53.5
12:09:00
55.6
12:00:00
96.5
11:59:00
98.4
12:04:00
87.9
12:05:00
89.4
time
temp
12:10:00
61.7
12:11:00
62.2
Full Sort: Sorting
SELECT temp
FROM weather
ORDER BY time ASC
12:01:00
12:02:00
12:03:00
12:06:00
88.3
89.3
88.0
91.1
12:07:00
52.4
12:08:00
53.5
12:09:00
55.6
12:00:00
96.5
11:59:00
98.4
12:04:00
87.9
12:05:00
89.4
time
temp
12:10:00
61.7
12:11:00
62.2
SELECT temp, date_trunc('hour', time)
FROM weather
ORDER BY date_trunc('hour', time) ASC
Works for monotonic functions too
Full Sort: TopK
SELECT temp
FROM weather
ORDER BY time
LIMIT 3
time
temp
11:59:00
98.4
12:00:00
96.5
12:01:00
88.3
12:02:00
89.3
12:03:00
88.0
12:04:00
87.9
12:05:00
89.4
12:06:00
91.1
12:07:00
52.4
12:08:00
53.5
12:09:00
55.6
12:10:00
61.7
12:11:00
62.2
Scan stops after K = 3 rows ✅
never read
Full Sort: Streaming / Non-Blocking Aggregation
SELECT time, city, avg(temp)
FROM weather
GROUP BY time, city
time
city
temp
12:00:00
Boston
75.4
12:00:00
Boston
76.2
12:00:00
Denver
84.9
12:00:00
SF
65.2
12:00:00
SF
63.8
12:01:00
Austin
94.3
12:01:00
NYC
71.8
12:01:00
NYC
72.4
12:02:00
Boston
74.7
12:02:00
Miami
88.1
12:02:00
Miami
87.5
Full Sort: Streaming / Non-Blocking Aggregation
SELECT time, city, avg(temp)
FROM weather
GROUP BY time, city
time
city
temp
12:00:00
Boston
75.4
12:00:00
Boston
76.2
12:00:00
Denver
84.9
12:00:00
SF
65.2
12:00:00
SF
63.8
12:01:00
Austin
94.3
12:01:00
NYC
71.8
12:01:00
NYC
72.4
12:02:00
Boston
74.7
12:02:00
Miami
88.1
12:02:00
Miami
87.5
① accumulate avg for current group
② new (time, city)! → group complete
…repeat for next group
③ emit avg row ✅
time
city
avg(temp)
12:00:00
Boston
75.8
Full Sort: Streaming / Non-Blocking Window Functions
SELECT region, time, temp,
first_value(temp) OVER (PARTITION BY region ORDER BY time)
FROM weather
region
time
temp
east
12:00:00
72.1
east
12:01:00
73.4
south
12:00:00
88.5
south
12:01:00
89.2
south
12:03:00
90.1
west
12:00:00
65.2
west
12:02:00
66.8
west
12:04:00
67.9
NOTE: some systems will first sort on PARTITION BY, ORDER BY columns and then use streaming windows computation
Full Sort: Streaming / Non-Blocking Window Functions
SELECT region, time, temp,
first_value(temp) OVER (PARTITION BY region ORDER BY time)
FROM weather
region
time
temp
east
12:00:00
72.1
east
12:01:00
73.4
south
12:00:00
88.5
south
12:01:00
89.2
south
12:03:00
90.1
west
12:00:00
65.2
west
12:02:00
66.8
west
12:04:00
67.9
① east rows are contiguous
② new region → new window
region
time
temp
first_temp
east
12:00:00
72.1
72.1
east
12:01:00
73.4
72.1
③ emit rows immediately ✅
Full Sort: Streaming / Non-Blocking Joins
SELECT time, sensor_id, name, value
FROM readings JOIN lookup USING (sensor_id)
sensor_id
name
s1
SF
s2
Boston
s3
Denver
sensor_id
time
value
s1
12:00:00
65.2
s1
12:01:00
64.1
s2
12:00:00
75.4
s2
12:01:00
76.2
s2
12:02:00
74.7
s3
12:00:00
84.9
s3
12:01:00
86.3
s3
12:02:00
85.7
NOTE: some plans will first sort the data by sensor_id then use merge join
Full Sort: Streaming / Non-Blocking Joins
SELECT time, sensor_id, name, value
FROM readings JOIN lookup USING (sensor_id)
sensor_id
name
s1
SF
s2
Boston
s3
Denver
sensor_id
time
value
s1
12:00:00
65.2
s1
12:01:00
64.1
s2
12:00:00
75.4
s2
12:01:00
76.2
s2
12:02:00
74.7
s3
12:00:00
84.9
s3
12:01:00
86.3
s3
12:02:00
85.7
① match rows on sensor_id
② new sensor_id (both sides)!
…repeat for next sensor_id
③ emit joined rows ✅
time
sensor_id
name
value
12:00:00
s1
SF
65.2
12:01:00
s1
SF
64.1
Partially Sorted Data
Full Sort vs Partial Sort
Partial Sort means sorted by a prefix of the optimal sort key
Full Sort
ORDER BY (time, city)
time
city
11:00:00
Austin
11:00:00
Boston
11:00:00
Chicago
11:01:00
Austin
11:01:00
Chicago
11:02:00
Boston
11:02:00
Chicago
11:03:00
Austin
Partial (prefix) Sort
ORDER BY (time)
time
city
11:00:00
Boston
11:00:00
Chicago
11:00:00
Austin
11:01:00
Chicago
11:01:00
Austin
11:02:00
Chicago
11:02:00
Boston
11:03:00
Austin
sort key
time
city
✅ ties sorted by city
❌ ties NOT sorted by city
prefix
Partial Sort: Streaming / Non-Blocking Sorting
SELECT temp
FROM weather
ORDER BY time, city ASC
time
city
temp
12:00:00
SF
65.2
12:00:00
Boston
75.4
12:00:00
Denver
84.9
12:01:00
NYC
71.8
12:01:00
Austin
94.3
12:02:00
Miami
88.1
12:02:00
Boston
76.2
12:02:00
Austin
95.6
12:03:00
SF
64.1
12:03:00
Denver
86.3
12:03:00
Boston
74.7
Partial Sort: Streaming / Non-Blocking Sorting
SELECT temp
FROM weather
ORDER BY time, city ASC
time
city
temp
12:00:00
SF
65.2
12:00:00
Boston
75.4
12:00:00
Denver
84.9
12:01:00
NYC
71.8
12:01:00
Austin
94.3
12:02:00
Miami
88.1
12:02:00
Boston
76.2
12:02:00
Austin
95.6
12:03:00
SF
64.1
12:03:00
Denver
86.3
12:03:00
Boston
74.7
① buffer rows with same time
② new time! → sort buffer by city
…repeat for next timestamp
time
city
12:00:00
Boston
75.4
12:00:00
Denver
84.9
12:00:00
SF
65.2
③ emit sorted rows ✅
temp
Partial Sort: Streaming / Non-Blocking Aggregation
SELECT time, city, avg(temp)
FROM weather
GROUP BY time, city
time
city
temp
12:00:00
SF
65.2
12:00:00
Boston
75.4
12:00:00
SF
63.8
12:00:00
Denver
84.9
12:00:00
Boston
76.2
12:01:00
NYC
71.8
12:01:00
Austin
94.3
12:01:00
NYC
72.4
12:02:00
Miami
88.1
12:02:00
Boston
74.7
12:02:00
Miami
87.5
Partial Sort: Streaming / Non-Blocking Aggregation
SELECT time, city, avg(temp)
FROM weather
GROUP BY time, city
time
city
temp
12:00:00
SF
65.2
12:00:00
Boston
75.4
12:00:00
SF
63.8
12:00:00
Denver
84.9
12:00:00
Boston
76.2
12:01:00
NYC
71.8
12:01:00
Austin
94.3
12:01:00
NYC
72.4
12:02:00
Miami
88.1
12:02:00
Boston
74.7
12:02:00
Miami
87.5
① accumulate avgs per (time,city)
② new time! → groups complete
…repeat for next timestamp
③ emit avg rows ✅
time
city
avg(temp)
12:00:00
Boston
75.8
12:00:00
Denver
84.9
12:00:00
SF
64.5
Partial Sort: Streaming / Non-Blocking
Window Functions
(multi-column) Joins
⇒ Exercise for the reader (I am already behind schedule)
Analysis: Equivalent Orderings
Needed:
Depends on
Analysis: Equivalent Orderings: Projection
Output ordering of project depends on the expressions
Scan
sorted: (time, city)
Project
exprs: date_trunc('hour', time)
① output order: (time, city)
② output order:
(date_trunc('hour',...), city)
Analysis: Equivalent Orderings: Expressions
time
city
11:58:00
Austin
11:58:00
Boston
11:59:00
Austin
12:00:00
Austin
12:01:00
Boston
date_trunc
('hour',time)
11:00:00
11:00:00
11:00:00
12:00:00
12:00:00
extract(minute
FROM time)
58
58
59
0
1
✅ sorted
❌ not sorted
Analysis: Equivalent Orderings: Functional Dependence
Scan
sorted: (time, city)
Filter
exprs: time = '12:05:00'
① output order: (time, city)
② output order:
(time, city)
(city, time)
time is known to be a single value: can be placed anywhere in the sort
Analysis: Equivalent Orderings: Joins
Merge Joining on a sorted key means output is sorted on both keys (as we know keys are equal)
* inner joins only — outer joins insert NULLs
MergeJoin
on: sensor_id
Scan lookup l
sorted: (l.sensor_id)
Scan readings r
sorted: (r.sensor_id, time)
② output order:
(r.sensor_id, time)
③ output order:
(l.sensor_id, r.sensor_id, time)
(r.sensor_id, l.sensor_id, time)
① output order:
(l.sensor_id)
Almost Sorted Data
Almost Sorted Data
time
temp
11:00:00
89.8
11:01:00
91.2
11:03:00
90.9
11:02:00
92.5
11:04:00
88.1
11:06:00
87.9
11:05:00
90.1
11:07:00
86.8
file_1.parquet
min: 11:00:00
max: 11:10:00
11:09:00
88.9
11:08:00
91.7
11:10:00
89.4
Pruning (Data Skipping)
SELECT avg(temp)
FROM weather
WHERE time BETWEEN '12:00:00' AND '12:05:00'
time
temp
11:00:00
89.8
…
…
11:59:00
98.4
file_1.parquet
min: 11:00:00
max: 11:59:00
time
temp
12:00:00
96.5
…
…
12:59:00
58.1
file_2.parquet
min: 12:00:00
max: 12:59:00
time
temp
13:00:00
63.2
…
…
13:59:00
60.9
file_3.parquet
min: 13:00:00
max: 13:59:00
Pruning (Data Skipping)
SELECT avg(temp)
FROM weather
WHERE time BETWEEN '12:00:00' AND '12:05:00'
time
temp
11:00:00
89.8
…
…
11:59:00
98.4
file_1.parquet
min: 11:00:00
max: 11:59:00
time
temp
12:00:00
96.5
…
…
12:59:00
58.1
file_2.parquet
min: 12:00:00
max: 12:59:00
time
temp
13:00:00
63.2
…
…
13:59:00
60.9
file_3.parquet
min: 13:00:00
max: 13:59:00
① compare filter to
file min/max stats
② prune files that
cannot match ❌
③ scan only file_2 ✅
❌ pruned
❌ pruned
Analysis: Range Analysis
time BETWEEN '12:00:00' AND '12:05:00'
file
time_min
time_max
file_1.parquet
11:00:00
11:59:00
file_2.parquet
12:00:00
12:59:00
file_3.parquet
13:00:00
13:59:00
Analysis: Range Analysis
time BETWEEN '12:00:00' AND '12:05:00'
file
time_min
time_max
file_1.parquet
11:00:00
11:59:00
file_2.parquet
12:00:00
12:59:00
file_3.parquet
13:00:00
13:59:00
time_max >= '12:00:00'
AND time_min <= '12:05:00'
① rewrite using min/max ranges*
② evaluate on each file’s ranges
✅
③ false → prune, scan only file_2 ✅
* there are more general frameworks that work for other types
❌
❌
Dynamic Filters
“This scheme was originally proposed in Goetz Graefe's paper… During execution, the scheme maintains a cutoff value and guarantees that records greater than this value will not appear in the TopK result set.” - http://mysql.taobao.org/monthly/2023/01/02/
* Chronis Y et al: External Merge Sort for Top-K Queries: Eager input filtering guided by histograms. SIGMOD 2020
* Graefe, A General and Efficient Algorithm for “Top” Queries, ICDE Workshops (DBRank) 2008
TopK Dynamic Filters
SELECT * FROM weather ORDER BY time LIMIT 3;
Scan
Select: * (all columns)
TopK
Limit: 3
Keys: time
Heap
①: decode all columns, filtering rows which would be filtered by heap
②: Examine all rows, keeping only the current smallest 3 in a heap
Filter: time < ??
③ Dynamic Filter is updated during execution as the values in the heap change
Heap min
Dynamic Filters: No Sort
Filter
time < ?
TopK
①: start
TopK
Filter
time < 11:07:00
②: after 4 rows
Heap: 11:02, 11:04, 11:07
TopK
Filter
time < 11:02:00
③: after 8 rows
Heap: 11:00, 11:01, 11:02
execution time
weather
time
temp
11:07:00
89.8
11:02:00
91.2
11:09:00
90.9
11:04:00
88.1
11:01:00
92.5
11:10:00
87.9
11:03:00
90.1
11:00:00
86.8
②
④: all skipped (≥11:02:00)
11:08:00
88.9
11:12:00
91.7
11:05:00
89.4
SELECT * FROM weather ORDER BY time LIMIT 3;
③
④
Dynamic Filters with Almost Sorted Data
Dynamic Filters with Almost Sorted Data
execution time
weather
time
temp
11:00:00
89.8
11:01:00
91.2
11:03:00
90.9
11:02:00
92.5
11:04:00
88.1
11:06:00
87.9
11:05:00
90.1
11:07:00
86.8
②
④: all skipped (≥11:02:00)
SELECT * FROM weather ORDER BY time LIMIT 3;
11:09:00
88.9
11:08:00
91.7
11:10:00
89.4
③
④
Filter
time < ?
TopK
①: start
TopK
Filter
time < 11:02:00
②: after 4 rows
Top 3 FOUND!
Heap: 11:00, 11:01, 11:02
TopK
Filter
time < 11:02:00
③: after 8 rows
Conclusion