Efficiently Merging Time Series in Kotlin with Sorted Sequences

Joining Kotlin sequences that are pre-sorted (like time series) in a lazy way

Imagine This:

You have two time series: one is a sequence of temperature readings, the other air pressure measurements. Each event in both sequences is keyed by a timestamp. Now, you want to combine temperature and pressure readings that occurred at the same time, allowing for duplicate or even missing timestamps. At first glance, this sounds easy to implement with Kotlin's standard collections, but there’s a catch…

The Challenge with Kotlin's Standard Library

Let’s say you start with an approach that first converts the time series into maps, keyed by their timestamps, then sorts the union of all keys and finally, by iterating over the keys, looks-up the corresponding readings in the maps and prints some combined message:

val temperatures = sequenceOf("10:00" to 16, "10:05" to 17, "10:10" to 18)
val pressures = sequenceOf("10:00" to 1020.0, "10:10" to 1018.0)

val temperatureMap = temperatures.toMap()
val pressureMap = pressures.toMap()

(temperatureMap.keys + pressureMap.keys).sorted().forEach {
    println("Time: $it, Temp: ${temperatureMap[it]}, Press: ${pressureMap[it]}")
}

//Output:
//Time: 10:00, Temp: 16, Press: 1020.0
//Time: 10:05, Temp: 17, Press: null
//Time: 10:10, Temp: 18, Press: 1018.0

This implementation works fine as long as your time series are small. But it will terribly fail when they consist of a huge (or even an infinite) number of elements. Our approach requires everything to be loaded in memory, which is not feasible for infinite or very large data sets. While Kotlin’s Sequence API is great for pipelines, it lacks built-in support for efficient operations on sorted sequences, like merging, joining, or deduplication based on keys.

Sorted Sequences to the Rescue

That’s why I wrote a small Kotlin library: SortedSequence. The idea is simple: if we know that a sequence is sorted according to its natural order or by some derived key, we can exploit that knowledge to perform efficient operations like merges and joins, without loading everything into memory and maintaining the streaming behavior. To make this safe, we have to explicitly assert the sort-order:

val seq1 = sequenceOf(1, 2, 3, 6).assertSorted()

val seq2 = sequenceOf("1a", "3c", "5b").assertSortedBy {
    parseInt(it[0].toString())
}

The result are sequences whose sort-order is checked while iterating, throwing an exception in case the sort-order is violated. This guarantee of being sorted allows for many operations to be implemented in an efficient way, like:

Join operations:

seq1.innerJoinByKey(seq2).forEach(::println)

//Output:
//(1, 1a)
//(3, 3c)

seq1.leftOuterJoinByKey(seq2).forEach(::println)

//Output:
//(1, 1a)
//(2, null)
//(3, 3c)
//(6, null)

Deduplication:

sequenceOf(1, 2, 2, 4, 4).assertSorted().distinct().forEach(::println)

//Output:
//1
//2
//4

Other operations that are supported include grouping, interleaving and zipping by key. All of these operations are streaming, so they scale to arbitrarily large inputs.

Solving Our Original Problem

Let’s go back to our original problem: combining the temperature and pressure readings. We now solve it using our SortedSequence:

val temperatures = sequenceOf("10:00" to 16, "10:05" to 17, "10:10" to 18)
val pressures = sequenceOf("10:00" to 1020.0, "10:10" to 1018.0)

val temperaturesSorted = temperatures.assertSorted()
val pressuresSorted = pressures.assertSorted()

temperaturesSorted.fullOuterJoinByKey(pressuresSorted).forEach { (time, values) -> 
    println("Time: $time, Temp: ${values.first}, Press: ${values.second}")
}

//Output:
//Time: 10:00, Temp: 16, Press: 1020.0
//Time: 10:05, Temp: 17, Press: null
//Time: 10:10, Temp: 18, Press: 1018.0

That’s it. Easy, readable, and memory-safe!

Conclusion

When working with large or infinite sorted sequences, Kotlin’s standard collections lack streaming support for certain operations. The SortedSequence library fills this gap by enabling efficient streaming operations like joins, and deduplications, while maintaining safety through runtime assertions on sort order.

➡️ Check it out on GitHub: timvanoijen/sorted-sequence Thanks for reading!