Eight rows per instruction: Spark's TIME type on Varka's 64-bit lane

Spark 4.1 gave SQL a TIME type: a time of day with no date attached, stored as nanoseconds since midnight in a 64-bit integer, with the usual functions over it - hour, minute, second, time_trunc, arithmetic with intervals, differences between two times. It arrived behind a flag and is on its way to being on by default.

On the development laptop - an AMD Ryzen AI 9 HX PRO 370, a Zen 5 core that runs 512-bit instructions on a 256-bit datapath, OpenJDK 25, one core - stock Spark 4.2 computes hour(t) over a cached column at 61.3 million rows a second. Varka, a research fork of Spark that compiles a projection into one vector loop, computes the same hour(t) on the same core at 1041.4 million rows a second: about seventeen times faster. The numbers that need a genuine 512-bit datapath are measured on one, a GitHub-hosted runner with an AMD EPYC 9V45, through a workflow anyone can dispatch; section 8 says which numbers those are. Every figure in this post comes from a committed results file that names its machine, JDK, row count and measured vector width, so each one can be checked without rerunning anything. This post is about where those seventeen come from. Not from a faster algorithm - the algorithm is a division - but from what the loop around it looks like, what it reads, and what it does not do per row.

The ideas come first and the numbers last, because the numbers only mean something once you can see the loop. If you already know how Spark's codegen works, section 3 is where Varka starts.

1. What Spark's codegen produces

Whole-stage code generation was the right idea in 2015 and it still is: instead of interpreting an expression tree per row, Spark writes the Java for a pipeline of operators, compiles it with Janino, and runs a tight loop that the JIT can inline through. It removed a virtual call per operator per row and won a lot.

What it produces is still a row loop. The generated code takes one row, applies every expression to it, writes one output row, and goes round again. Every column value is read out of an UnsafeRow, every intermediate is a local, and anything that needs an object - a LocalTime, a Decimal, a UTF8String - gets a new one per row.

Figure 1. Left: the generated Java takes rows one at a time and uses one lane of a core that has eight 64-bit lanes to offer. Right: Varka's emitted loop takes eight values of the column per instruction.

The core underneath has moved on. A server core today does 512 bits of integer work per instruction: eight 64-bit values, sixteen 32-bit ones. A row loop uses one lane of that and leaves the other seven idle, and no amount of JIT cleverness recovers them, because the loop's shape - one row, many columns, then the next row - is the wrong way round for the hardware. The vector unit wants one column, many rows.

2. What hour(t) costs

Take the smallest example. A TIME is a long: nanoseconds since midnight. The hour is that long divided by 3 600 000 000 000. Here is how stock Spark gets there, from DateTimeUtils:

def getHoursOfTime(nanos: Long): Int = LocalTime.ofNanoOfDay(nanos).getHour

Figure 2. Stock Spark builds a LocalTime - four fields, one allocation - to read one of the fields back. The lane divides.

That is a perfectly reasonable line of Scala, and it is what a row engine almost has to write: java.time is the correct library and getHour is the correct method. The cost is that LocalTime.ofNanoOfDay does three divisions to fill four fields, allocates an object to hold them, and the caller reads one field and drops the object. Escape analysis sometimes removes the allocation and sometimes does not, and the three divisions stay either way. Measured over five hundred million cached rows, hour(t) costs stock Spark 16.3 nanoseconds a row on this machine, and minute(t) and second(t) 16.4 each, because they are the same object built for a different field.

Varka's lane does one 64-bit division and stores the quotient. It costs 1.0 nanoseconds a row, and most of that is the memory traffic - an eight-byte read and a four-byte write per row - rather than the division.

3. A batch is columns, not rows

The first thing Varka changes is what the loop reads. Spark already carries columns in the JVM - a cached table lives in ColumnarBatches - and Varka adds a cache serializer that writes those batches as Arrow. That is where the kernels get their columns today, and it is how every benchmark in this post is set up: the table is cached, so the measurement times the loop rather than a scan. An Arrow column is two buffers, the values back to back and a validity bitmap with one bit per row, and nothing else.

The cache is the first source, not the only one it can have. A columnar datasource that hands Arrow batches straight out of a file - Parquet is the obvious one, since its pages are already columns - would feed the same kernels with no cache step in front, and that is the natural next reader for this engine. Nothing below depends on where the batch came from; it depends only on the batch being Arrow.

Figure 3. The varka_times table the benchmarks use: two TIME columns, two day-time intervals, two bigints, eight bytes a row each, plus a bit per row for nulls. The loop reads the buffers where Arrow allocated them.

Varka maps each buffer to a Panama MemorySegment over the address Arrow already owns - no copy, no ByteBuffer, no object per value - and the emitted loop loads eight lanes from it at a time with the Vector API. The validity bitmap is loaded as bits and becomes a VectorMask, so a null is a lane that is masked out rather than a branch.

This is the point where a native engine - Velox, DataFusion, Comet, Gluten - crosses into C++ over JNI, and gets a very good vector engine on the other side. Varka stays in the JVM on purpose: one jar for every CPU, one memory and crash domain, ordinary profilers and stack traces, and eventually a user's own vectorised function fused into the same loop as the built-ins instead of rows handed back across a boundary. Whether the JVM can do the vector part well enough is the experiment, and it is what the numbers at the end are for.

4. One class per projection, and a trapdoor

The second change is what the loop is. Varka does not interpret a vector expression tree, and it does not call a library of kernels either - a call per operator per batch is cheap, but the calls are megamorphic and the JIT cannot see across them. Instead the compiler turns the Catalyst expressions of a projection into a small vector IR, and an emitter writes a Java class for that IR with JDK 25's Class-File API: one class per projection shape.

Inside that class the loop is not always a single method. Outputs that share work stay together, and outputs that share nothing are split into sibling methods, because C2's compile time grows steeply with the number of vector operations in a method and a wide method is both slower to compile and slower to run. Sibling methods, not longer methods, is the rule the emitter budgets against.

Figure 4. Above: a projection becomes an IR, the IR becomes bytecode, the JIT sees one loop with monomorphic call sites. Below: batch by batch at run time, with the trapdoor to the row engine under the kernel.

Two consequences are worth knowing about. The class is named after its plan node - its SourceFile attribute reads like Project#12 - and carries a custom attribute with the IR it computes, so a profiler, a flame graph or a heap dump names the query's operator with no mapping table. And the class is loaded by the task that needs it and unloaded with it, so a long-running executor does not accumulate a codegen cache that only grows.

The trapdoor is the contract that makes the rest possible. Anything the compiler cannot lower declines per expression at plan time, and the stock row path computes that column while the fusable ones still fuse. Anything the emitted kernel cannot compute for a batch - a TIME plus an interval that would cross midnight, which is an error in Spark, or any failure at all in the kernel - is refused, and the row engine computes that batch instead. A query on Varka can be slower than it should be; it cannot be wrong or fail because of Varka. That rule is what lets the engine grow one expression family at a time in front of users rather than behind them.

5. The 64-bit lane, and why a division is the expensive thing

TIME is the third type on Varka's 64-bit lane, after bigint and day-time intervals, and it is the one that made the lane interesting, because nearly everything you do to a time of day is a division by a constant: the hour is a division by 3.6e12, the minute by 6e10 and 60, time_trunc('MINUTE', t) a division and a multiplication, time_diff a subtraction and a division. On the 32-bit lane a division by a constant is a multiply-high and a shift - a few cheap instructions. On the 64-bit lane there is no multiply-high in the Vector API, and a 64-bit product overflows anyway, so the lane goes through doubles.

Figure 5. With AVX-512 the machine converts eight longs to eight doubles in one instruction, divides, and converts back: three vector operations. With AVX2 there is no such conversion, so the lane reads the bits of the long as a double through an identity and back again: fourteen.

A double holds every integer below 2^53 exactly, so this form's quotient is exact for every dividend below 2^53 - and a day is 8.64e13 nanoseconds, under 2^47, with room to spare. So the conversion form is exact, and on a machine with AVX-512 it is a convert, a divide and a convert - three vector operations, because a 64-bit lane and a double lane are the same width, so there are no halves to split and rejoin the way the 32-bit lane has to.

Most laptops and a good share of cloud machines have AVX2 and not AVX-512, and AVX2 has no instruction that turns a vector of longs into a vector of doubles. HotSpot then compiles the conversion as eight scalar conversions, which is worse than not vectorising. For those machines Varka emits a second form built on a bit trick: for 0 <= u < 2^52, the bit pattern of u OR'd with 0x4330000000000000 read as a double is exactly 2^52 + u, so a subtraction recovers u as a double with no conversion instruction at all, and the same identity run backwards turns the quotient into a long. The floor that Java's truncating division needs is built by hand - the Vector API has no lanewise floor either - and the sign is handled by dividing the magnitude and negating the lanes that need it afterwards. Fourteen operations against three, all of them vector operations, all of them instructions AVX2 has.

Both forms are exact for every dividend under 2^52, and the emitter carries that bound as an obligation on every 64-bit division it compiles, checks it against what it knows of the dividend, and a test drives both forms through the JVM at the bound to confirm that each fails exactly the way its arithmetic says it should above it. The lowering is chosen per machine, at emission, from what the JVM reports it can do, and the shape of the compiled kernel is part of the cache key, so a jar moved between machines recompiles rather than misbehaves.

6. One loop for the whole projection

The third change is the one that compounds. A projection usually has several outputs, and they usually share work: hour(t + dt) and time_trunc('MINUTE', t + dt) both need t + dt, and both need it guarded against leaving the day. Spark's codegen already spots that: subexpression elimination is on by default, and the generated row loop computes the sum once per row into a local. A kernel-library engine has a harder time of it - each operator is a call over a whole batch, so the shared value is computed once per batch but materialised into memory and read back. Varka compiles the projection as one graph, finds the shared subtrees, and emits a loop in which the shared value is computed once per eight rows and stays in a register between its two uses. The saving is not that the work happens once; it is that it happens once for eight rows and never leaves the register file.

Figure 6. Left: the expression graph of two outputs over the same guarded sum. Right: the loop the emitter writes for it - one load per input column, the shared sum once, one store per output.

This is where the row engine's overhead stops being amortised and starts being multiplied: every link of a chain costs stock Spark another virtual call, another boxed intermediate or another object, while the fused loop pays one load per input column and one store per output whatever the depth. It is also why a single-expression benchmark understates the engine. The TIME surface below measures one call per row and reads 13x to 38x; the chains, which compose the calls, are where the fused loop shows what it is worth, and they are the number the closing section is built around.

7. How it is measured

A "17x" is worth exactly as much as the method behind it, so briefly: what was compared, on what, and what would have made a run invalid.

Figure 7. Two stock releases on two JDKs, the fork with the engine off, and the fork with it on, over the same cached rows, timed by one benchmark jar that depends on no part of the fork.

Four arms. Every table compares stock Spark 4.2.0 on JDK 17 and on JDK 25, this fork with the engine off, and this fork with Varka on. The engine-off arm is the control that separates the engine from the fork: the fork tracks Spark master, and on t + dt master's own row path already runs at 52.3 million rows a second where 4.2.0 runs at 33.5. So every ratio is against stock, and where the two baselines disagree by more than a fifth the ratio against the fork's own row engine is printed beside it. Runs are one core over cached Arrow rows, with at least five iterations over two-second windows, recording wall time and executor time separately.

Two tables, two machines, and the reason is arithmetic. The chains - composed expressions, several operations deep - are measured on GitHub Actions, on a runner whose vector datapath the measuring job itself proves is 512 bits wide. They are the numbers this post leads with, and anyone can reproduce them by dispatching the same workflow. The surface - one row per expression, the coverage document - stays on the development laptop and is labelled as such, because it cannot be measured honestly on a runner at all: its lightest entries run at 0.8 nanoseconds a row, a cloud runner's per-iteration constant is about 36 milliseconds, and keeping that constant under 5% means the work has to last at least 720 milliseconds - some nine hundred million rows, which at this table's 49 bytes a row is about 41 GiB of cache on a machine with 15 GiB of memory. A surface row therefore says what fuses and roughly what it is worth on a developer's machine; a chain row is the claim.

Figure 8. Six guards between a run and a committed file, each one the memory of a number that was wrong once.

Six gates. On a development machine a run is refused unless the canary - three fixed loops - reads within a few percent of that host's committed baseline, which is how a run knows the machine is in its measured state. It cannot run on a cloud runner, whose hostname is new every dispatch and which therefore has no baseline to compare against, so the runner files record canary: OFF and a reader can see exactly which of these guards was in force. The rest apply everywhere. The datapath probe records how wide the vector unit really is (the 512:256 ratio reads 1.14 on this laptop, 1.35 on the Intel Xeons of the CI pool and 2.01 on the EPYC 9V45, and only the last is full width); and the table stayed resident, because one that spills recomputes every iteration and passes every other rule with a better number. A run fails if any entry expected to fuse planned without a Varka node or if a single batch fell through the trapdoor, since a blend of kernel and row-engine batches looks like a kernel rate; and it fails the fixed-share rule if the job's constant - about 15 ms on the laptop, 36 ms on a runner - exceeds 5% of any Varka row's wall time (one row sits at 5.9%, and the bound was lifted to 6% for it, in writing). Finally the band: twelve repeated runs give each case a tier, and no row is quoted without one. After all that, a check that every number with decimals in the documents traces to a committed file, or the build fails.

Figure 9. The GitHub-hosted pool as eighteen dispatches saw it, and the workflow that measures on the one machine in it whose datapath is genuinely 512 bits wide - proven by the measuring job itself, and written into the file.

Where the 512-bit numbers come from. Not from this laptop, whose Zen 5 core double-pumps. The chains are measured by a GitHub Actions workflow, varka-surface-benchmark.yml, that anyone can dispatch: a build job caches the jars for the commit on whatever runner it gets, and a measure job starts by running the datapath probe on its own VM - a probe in any other job would be describing a different machine, which is a mistake the previous milestone made once - and stops within a minute unless the ratio is near 2. Across the pool that is one dispatch in about eighteen, an AMD EPYC 9V45; the Intel Xeons carry every AVX-512 flag and still read 1.35, because they issue 512-bit integer operations on fewer ports. A hit runs the four arms in about an hour and commits a file whose header carries the CPU and the probe's reading from the same VM, so the claim and its evidence cannot come apart. Milestone 4's date chains were measured this way - 9.4x to 14.2x against stock, median 10.2x - and the TIME chains of section 8 will be too.

8. The numbers

The headline is the chains, on the machine that earns the word full-width. Twelve chained TIME expressions, three to five operations deep, over a hundred million cached rows on one core of an AMD EPYC 9V45 whose measuring job proved its own datapath at 1.97. Varka runs them at 3.7 to 5.5 nanoseconds a row - 182 to 273 million rows a second - against stock Spark 4.2.0's 88 to 197 nanoseconds a row on the same machine and the same rows.