rust 57 lines · 8 steps

K-way merge of sorted logs in Rust

Merge many timestamp-sorted log files into one ordered stream using a min-heap that holds one line per source.

Explained by highlit
Intermediate binary-heap k-way-merge streaming-io custom-ordering
1use std::cmp::{Ordering, Reverse};
2use std::collections::BinaryHeap;
3use std::fs::File;
4use std::io::{self, BufRead, BufReader, BufWriter, Lines, Write};
5 
6struct Entry {
7 timestamp: i64,
8 line: String,
9 source: usize,
10}
11 
12impl PartialEq for Entry {
13 fn eq(&self, other: &Self) -> bool {
14 self.timestamp == other.timestamp
15 }
16}
17impl Eq for Entry {}
18impl PartialOrd for Entry {
19 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
20 Some(self.cmp(other))
21 }
22}
23impl Ord for Entry {
24 fn cmp(&self, other: &Self) -> Ordering {
25 self.timestamp.cmp(&other.timestamp)
26 }
27}
28 
29fn parse_ts(line: &str) -> i64 {
30 line.split_whitespace()
31 .next()
32 .and_then(|s| s.parse().ok())
33 .unwrap_or(i64::MAX)
34}
35 
36pub fn merge_logs(paths: &[&str], out: &str) -> io::Result<()> {
37 let mut readers: Vec<Lines<BufReader<File>>> = paths
38 .iter()
39 .map(|p| File::open(p).map(|f| BufReader::new(f).lines()))
40 .collect::<Result<_, _>>()?;
41 
42 let mut heap = BinaryHeap::new();
43 for (source, reader) in readers.iter_mut().enumerate() {
44 if let Some(line) = reader.next().transpose()? {
45 heap.push(Reverse(Entry { timestamp: parse_ts(&line), line, source }));
46 }
47 }
48 
49 let mut writer = BufWriter::new(File::create(out)?);
50 while let Some(Reverse(entry)) = heap.pop() {
51 writeln!(writer, "{}", entry.line)?;
52 if let Some(line) = readers[entry.source].next().transpose()? {
53 heap.push(Reverse(Entry { timestamp: parse_ts(&line), line, source: entry.source }));
54 }
55 }
56 writer.flush()
57}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A min-heap holding one item per input turns a k-way merge into repeated pop-and-refill steps with O(log k) per line.
  2. 2Implementing Ord on your struct lets standard collections order records by whatever field matters.
  3. 3Streaming lines through buffered readers and a heap keeps memory bounded to k entries no matter how large the files are.

Related explainers

rust 
use serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
rust 
use std::f64::consts::PI;
 
#[derive(Debug, Clone, Copy)]
pub struct LatLng {

Building geographic bounding boxes in Rust

geospatial value-types option
Intermediate 7 steps
rust 
use chrono::{Duration, NaiveDate};
 
#[derive(Debug)]
pub struct DateRange {

Parsing and iterating date ranges in Rust

error-handling iterators parsing
Intermediate 7 steps
rust 
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
 

Building a counting semaphore in Rust

concurrency synchronization condition-variable
Advanced 9 steps
rust 
use axum::{
    async_trait,
    extract::{rejection::JsonRejection, FromRequest, Request},
    http::StatusCode,

A validated JSON extractor in Axum

extractors validation error-handling
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

K-way merge of sorted logs in Rust — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code