OLAP – Phase 10 Sorting, Parallel Execution, REPL, and Server
The database can parse SQL, plan queries, join tables, and aggregate results. But it’s missing four things that make it usable: ORDER BY (sorting results), parallel execution (using multiple CPU cores), an interactive REPL for exploring data, and a TCP server for remote connections.
This phase ties everything together into a complete, runnable OLAP engine.
In DuckDB, sorting is in PhysicalOrder (src/execution/operator/order/physical_order.cpp), parallelism uses a morsel-driven pipeline model (src/parallel/), and the CLI is the duckdb shell.
This is the final phase of the OLAP series:
- Phase 10: Sorting, parallel execution, REPL, and server
Full Source Code
The code referenced in this post can be found in https://gitlab.com/kimserey.lam/olap-learn.
Sort Operator
The SortOperator is a pipeline breaker — it must consume all input before producing any output (you can’t know the first result row until you’ve seen the last input row):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
type SortKey struct {
Expression Expression
Direction SortDirection // SortAsc or SortDesc
Nulls NullOrder // NullsFirst, NullsLast, or NullsDefault
}
type SortOperator struct {
child PhysicalOperator
sortKeys []SortKey
rows [][]any
types []vector.LogicalType
resultIdx int
done bool
}
Three-Phase Execution
- Consume: materialize all input rows plus sort key values
- Sort: sort the materialized rows using Go’s
sort.SliceStable - Emit: yield sorted rows as DataChunks
1
2
3
4
5
6
7
8
func (s *SortOperator) Next() *vector.DataChunk {
if !s.done {
s.consumeAll()
s.sortRows()
s.done = true
}
return s.emitNextChunk()
}
Multi-Key Sorting
Sorting by multiple keys works by comparing the first key, then breaking ties with the second key, and so on:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
sort.SliceStable(s.rows, func(i, j int) bool {
for k, sk := range s.sortKeys {
keyIdx := numCols + k
a := s.rows[i][keyIdx]
b := s.rows[j][keyIdx]
aNull := a == nil
bNull := b == nil
if aNull && bNull {
continue
}
if aNull || bNull {
return nullComesFirst(aNull, sk.Direction, sk.Nulls)
}
cmp := compareAny(a, b)
if cmp == 0 {
continue // tie — check next key
}
if sk.Direction == SortDesc {
return cmp > 0
}
return cmp < 0
}
return false
})
NULL Ordering
SQL supports NULLS FIRST and NULLS LAST. The default is database-specific — we follow the convention where NULLs sort last for ASC and first for DESC:
1
2
3
4
5
6
7
8
9
10
11
12
13
func nullComesFirst(aIsNull bool, dir SortDirection, nulls NullOrder) bool {
switch nulls {
case NullsFirst:
return aIsNull
case NullsLast:
return !aIsNull
default:
if dir == SortAsc {
return !aIsNull // non-null comes first in ASC
}
return aIsNull // null comes first in DESC
}
}
Parallel Execution
OLAP queries process large amounts of data — using multiple CPU cores is essential. We implement morsel-driven parallelism, where row groups are divided into “morsels” that worker goroutines process independently.
Morsel Source
The morsel source distributes row groups to workers:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
type MorselSource struct {
table *catalog.Table
colIdxs []int
mu sync.Mutex
nextRG int
}
func (m *MorselSource) NextMorsel() *vector.DataChunk {
m.mu.Lock()
defer m.mu.Unlock()
for m.nextRG < len(m.table.RowGroups) {
rg := m.table.RowGroups[m.nextRG]
m.nextRG++
chunk := rg.ToDataChunk(m.colIdxs)
if chunk.Count() > 0 {
return chunk
}
}
return nil
}
Each worker calls NextMorsel() to get the next chunk to process. The mutex ensures no two workers process the same row group.
Parallel Aggregation
For aggregation queries, each worker maintains a local hash table. After all workers finish, the local results are merged:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
func (p *ParallelAggregator) Execute() *execution.HashAggregateOperator {
var wg sync.WaitGroup
var results []localResult
for i := 0; i < p.numWorkers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
localHT := execution.NewGroupByHashTable(funcs)
for {
chunk := p.source.NextMorsel()
if chunk == nil {
break
}
// apply filter if any
// process chunk into local hash table
processChunkIntoHT(chunk, p.groupByExprs, p.aggregates, localHT)
}
// collect local results
localHT.Scan(func(keys []any, states []execution.AggregateState) {
lr.keys = append(lr.keys, keys)
lr.states = append(lr.states, states)
})
mu.Lock()
results = append(results, lr)
mu.Unlock()
}()
}
wg.Wait()
// merge local results with a final HashAggregateOperator
// ...
}
The merge phase creates a new HashAggregateOperator that combines the partial results from all workers. For SUM, this means summing the local sums. For COUNT, summing the local counts.
REPL
The interactive REPL provides a command-line interface for exploring data:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
type REPL struct {
catalog *catalog.Catalog
out io.Writer
}
func (r *REPL) Run(in io.Reader) {
scanner := bufio.NewScanner(in)
fmt.Fprint(r.out, "olapdb> ")
var buf strings.Builder
for scanner.Scan() {
line := scanner.Text()
if buf.Len() == 0 && strings.HasPrefix(line, "\\") {
r.handleMeta(line) // \dt, \d tablename
continue
}
buf.WriteString(line + " ")
if strings.Contains(line, ";") {
r.execute(strings.TrimSpace(buf.String()))
buf.Reset()
fmt.Fprint(r.out, "olapdb> ")
} else {
fmt.Fprint(r.out, " ...> ")
}
}
}
SQL Execution Pipeline
Each SQL statement flows through the full pipeline:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
func (r *REPL) execSelect(sql string, start time.Time) {
stmt, _ := parser.Parse(sql)
b := planner.NewBinder(r.catalog)
logical, _ := b.Bind(stmt)
logical = planner.Optimize(logical)
pp := physical.NewPhysicalPlanner(r.catalog)
phys, _ := pp.Plan(logical)
phys.Init()
defer phys.Close()
// collect results, print as ASCII table
for {
chunk := phys.Next()
if chunk == nil {
break
}
// format rows
}
elapsed := time.Since(start)
printTable(r.out, headers, rows)
fmt.Fprintf(r.out, "(%d rows, %s)\n", len(rows), elapsed)
}
Meta-Commands
\dt— list all tables\d tablename— describe table schema (column names and types)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
func (r *REPL) handleMeta(cmd string) {
switch {
case cmd == "\\dt":
names := r.catalog.TableNames()
for _, name := range names {
fmt.Fprintln(r.out, " "+name)
}
case strings.HasPrefix(cmd, "\\d "):
tableName := strings.TrimSpace(strings.TrimPrefix(cmd, "\\d "))
table, _ := r.catalog.GetTable(tableName)
fmt.Fprintf(r.out, "Table: %s\n", table.Name)
for _, col := range table.Columns {
fmt.Fprintf(r.out, " %-20s %s\n", col.Name, typeName(col.Type))
}
}
}
TCP Server
The server accepts TCP connections and runs SQL queries for remote clients:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
type Server struct {
catalog *catalog.Catalog
listener net.Listener
}
func (s *Server) ListenAndServe(addr string) error {
ln, err := net.Listen("tcp", addr)
if err != nil {
return err
}
s.listener = ln
for {
conn, err := ln.Accept()
if err != nil {
return err
}
go s.handleConn(conn)
}
}
func (s *Server) handleConn(conn net.Conn) {
defer conn.Close()
scanner := bufio.NewScanner(conn)
r := repl.New(s.catalog, conn)
var buf strings.Builder
for scanner.Scan() {
line := scanner.Text()
if line == "quit" || line == "exit" {
fmt.Fprintln(conn, "Bye!")
return
}
buf.WriteString(line + " ")
if strings.Contains(line, ";") {
r.ExecuteSQL(strings.TrimSpace(buf.String()))
buf.Reset()
}
}
}
Each connection gets its own goroutine. The server reuses the REPL’s execution logic, directing output to the TCP connection instead of stdout.
Entry Point
The main program supports both REPL and server modes:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func main() {
serverMode := flag.Bool("server", false, "run as TCP server")
port := flag.Int("port", 5433, "server port")
dataDir := flag.String("data", "", "data directory for persistence")
flag.Parse()
var c *catalog.Catalog
if *dataDir != "" {
c, _ = catalog.NewCatalogWithDir(*dataDir)
} else {
c = catalog.NewCatalog()
}
if *serverMode {
addr := fmt.Sprintf(":%d", *port)
s := server.New(c)
s.ListenAndServe(addr)
} else {
r := repl.New(c, os.Stdout)
r.Run(os.Stdin)
}
}
Usage:
1
2
3
olapdb # Interactive REPL
olapdb --server --port 5433 # TCP server mode
olapdb --data ./data # Persistent storage
Summary
This final phase completes the OLAP engine with the missing pieces: ORDER BY via a sort operator (with multi-key comparison and NULL ordering), parallel execution using morsel-driven parallelism (each worker processes row groups independently with local aggregation), an interactive REPL with meta-commands, and a TCP server for remote connections. The result is a complete, runnable columnar database engine — from SQL text to query results — built from scratch in Go.
The full architecture:
1
2
3
4
5
6
7
8
9
10
11
SQL String
↓
[Parser] → AST
↓
[Binder] → Logical Plan
↓
[Optimizer] → Optimized Plan
↓
[Physical Planner] → Operator Pipeline
↓
[Executor] → DataChunks → Results
Each phase built one layer of this stack, from the foundational Vector and DataChunk (Phase 1) through storage, compression, cataloging, expressions, aggregation, joins, parsing, planning, and finally the interface layer.
