The 2,400-Line Bytecode Engine: Inside LTSeq's Core Algorithms

Summary: LTSeq's edge doesn't come from DataFusion's general query capabilities. It comes from three algorithms built for ordered sequences: a streaming linear scan for single-pass boundary detection, direct Parquet parallel scanning that bypasses DataFusion, and vectorized pattern matching for single-pass funnel analysis. This article covers the design and implementation of all three.

1. Background: the DataFusion bottleneck

For session segmentation like:

python

events.group_ordered(lambda r:
    (r.userid != r.userid.shift(1)) |
    (r.eventtime - r.eventtime.shift(1) > 1800))

The standard DataFusion path requires three full-table scans:

LAG window: compute userid.shift(1) and eventtime.shift(1)
Boundary marking: IS DISTINCT FROM plus CASE WHEN
Cumulative sum: SUM(boundary) OVER (ORDER BY ...) to produce group_id

Each pass reads and writes the entire table. For 100 million rows, that means three Arrow memory allocations and deallocations.

LTSeq's linear scan engine merges those three passes into a single O(N) scan.

2. The linear scan engine (`src/ops/linear_scan.rs`, 2,409 lines)

2.1 Eligibility check: who gets the fast path?

Not all expressions qualify for linear scanning. can_linear_scan() (line 55) performs two checks:

rust

pub fn can_linear_scan(expr: &PyExpr) -> bool {
    is_supported_expr(expr)  // Only contains supported operations?
        && contains_shift(expr)  // Contains at least one shift(1)?
}

The supported operation subset:

Operation	Evaluation
`Column("x")`	Read column value at current row
`shift(1)` on Column	Read column value at previous row
`is_null(expr)`	Check if evaluated value is NULL
`BinOp(Ne/Eq/Gt/Lt/Ge/Le)`	Compare two values
`BinOp(Add/Sub/Mul/Div)`	Arithmetic
`BinOp(Or/And)`	Logical combination
`Literal`	Constant value
`UnaryOp(Not)`	Logical negation

If the expression doesn't contain shift(1), it falls back to DataFusion's standard IS DISTINCT FROM LAG path. A simple column comparison doesn't need the linear scan engine.

2.2 Runtime value system

The linear scan engine has its own value type system, independent of DataFusion (src/ops/linear_scan.rs:149):

rust

enum Value {
    Null,
    Bool(bool),
    Int64(i64),
    Float64(f64),
    Str(String),
}

A critical design decision: timestamps are zero-copy reinterpreted as Int64. Whether second-level, millisecond-level, or microsecond-level timestamps, the underlying representation is an i64 integer. Difference calculations use plain integer subtraction, avoiding timestamp format conversion overhead:

rust

DataType::Timestamp(_, _) => {
    // Timestamps stored as i64 internally, read directly for comparison and arithmetic
    if let Some(arr) = array.as_any().downcast_ref::<TimestampSecondArray>() {
        Value::Int64(arr.value(row))
    } else if let Some(arr) = array.as_any().downcast_ref::<TimestampMicrosecondArray>() {
        Value::Int64(arr.value(row))
    }
    // ...
}

Cross-type numeric comparison is also handled elegantly:

rust

impl PartialEq for Value {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Value::Int64(a), Value::Int64(b)) => a == b,
            // Cross-type: promote to f64
            (Value::Int64(a), Value::Float64(b)) => (*a as f64) == *b,
            (Value::Float64(a), Value::Int64(b)) => *a == (*b as f64),
            // ...
        }
    }
}

2.3 TypedColumn: compile-time type elimination

Frequent downcast_ref calls are a common performance bottleneck in Arrow programming. The TypedColumn enum resolves types once at the batch level, making subsequent row-level access zero-overhead (src/ops/linear_scan.rs:301):

rust

enum TypedColumn {
    Int64(Arc<Int64Array>),
    UInt64(Arc<UInt64Array>),
    Int32(Arc<Int32Array>),
    UInt32(Arc<UInt32Array>),
    Float64(Arc<Float64Array>),
    Bool(Arc<BooleanArray>),
    Generic(ArrayRef),  // Fallback to dynamic dispatch for unsupported types
}

impl TypedColumn {
    fn from_array(array: &ArrayRef) -> Self {
        match array.data_type() {
            DataType::Int64 => TypedColumn::Int64(
                Arc::new(array.as_any().downcast_ref::<Int64Array>().unwrap().clone())
            ),
            // ...
        }
    }

    fn read_value(&self, row: usize) -> Value {
        match self {
            TypedColumn::Int64(arr) => {
                if arr.is_null(row) { Value::Null }
                else { Value::Int64(arr.value(row)) }  // Direct index, no type dispatch
            }
            // ...
        }
    }
}

Each batch resolves types once per column, then reads all rows through the typed reference. For a 100M-row x 2-column scenario, downcast_ref calls drop from 200 million to approximately 12,000 (assuming 16,384 rows per batch).

2.4 Fused streaming evaluation

This is the most critical optimization in the linear scan engine. streaming_fuse_eval pattern-matches common predicate shapes and fuses multi-step evaluation into a single step.

For the predicate r.eventtime - r.eventtime.shift(1) > 1800 ("adjacent row difference comparison"), standard evaluation requires:

Read eventtime[i]
Read eventtime[i-1]
Compute the difference
Compare against 1800

Fused evaluation merges these 4 steps into a tight loop:

rust

// Pseudocode: fused evaluation path
for i in 0..num_rows {
    let curr = vals[i];
    let prev = if i == 0 { prev_from_last_batch } else { vals[i-1] };
    boundary[i] = curr.wrapping_sub(prev) > threshold;
}

There are no intermediate array allocations: no new array for the shift(1) result, none for the difference, none for the comparison result. One loop handles everything.

For compound predicates like (r.userid != r.userid.shift(1)) | (r.eventtime - r.eventtime.shift(1) > 1800), the fused evaluator handles both conditions in the same loop iteration:

rust

// Pseudocode: compound fused evaluation
for i in 0..num_rows {
    let uid_curr = uid_vals[i];
    let uid_prev = if i == 0 { state.prev_uid } else { uid_vals[i-1] };
    let time_curr = time_vals[i];
    let time_prev = if i == 0 { state.prev_time } else { time_vals[i-1] };

    boundary[i] = (uid_curr != uid_prev) || (time_curr - time_prev > 1800);
}

2.5 Cross-batch state: StreamState

Arrow's streaming processing operates in RecordBatch units, but shift(1) needs the previous row's value, which might sit in the last row of the preceding batch. StreamState handles that:

rust

pub struct StreamState {
    prev_values: HashMap<String, Value>,  // Column values from the last row of the previous batch
    current_gid: i64,                      // Current group_id counter
}

After each batch is processed, the last row's key column values are saved to prev_values. When the next batch starts, these values serve as the shift(1) result for row 0.

2.6 NULL semantics: conservative by design

The linear scan engine handles NULLs conservatively, preferring too many groups over too few:

NULL in Ne comparison -> result is true (treated as boundary, creates new group)
Final evaluation result is NULL -> treated as true (creates new group)

This semantic guarantees that NULL values are never erroneously merged into the preceding group. It's the safe default for session segmentation: if you don't know whether a row belongs to the same session, treat it as a new session.

3. Direct Parquet parallel engine (`src/ops/parallel_scan.rs`, 984 lines)

3.1 Why bypass DataFusion?

DataFusion supports parallel execution, but its SortPreservingMergeExec introduces merge overhead in multi-partition mode. For pre-sorted Parquet files, data is ordered both within row groups and across them, so no merge is needed.

LTSeq's direct Parquet engine uses Rayon to read row groups in parallel, avoiding DataFusion's plan construction and sort-merge overhead.

3.2 Strategy one: sequential streaming (for group_ordered)

direct_streaming_group_ordered() (parallel_scan.rs:42) execution flow:

rust

pub fn direct_streaming_group_ordered(
    table: &LTSeqTable, predicate: &PyExpr, parquet_path: &str,
) -> PyResult<LTSeqTable> {
    // 1. Extract columns referenced by the predicate
    let needed_cols = extract_referenced_columns(predicate);

    // 2. Build Parquet column projection (only read needed columns)
    let projection_mask = ProjectionMask::roots(schema_descr, proj_indices);

    // 3. Sequential read + fused streaming evaluation
    let mut state = StreamState::new();
    let mut all_boundaries = Vec::with_capacity(total_rows);

    for batch in reader {
        let boundary_flags = streaming_fuse_eval(predicate, &batch, &mut state);
        all_boundaries.extend(boundary_flags);
    }

    // 4. Single-pass construction of group_id / group_count / rn
    build_metadata_table(all_boundaries, ...)
}

Key optimizations:

Column projection. Only predicate-referenced columns are read, which cuts I/O.
Streaming. The whole file never has to be held in memory.
Fused evaluation. Boundary detection happens while the data is being read.

3.3 Strategy two: Rayon parallel (for pattern matching)

parallel_pattern_match_count() uses Rayon for row-group-level parallel processing:

rust

// Pseudocode
let counts: Vec<usize> = row_groups
    .into_par_iter()
    .map(|rg_idx| {
        // Each task independently opens the Parquet file
        let file = File::open(parquet_path)?;
        let reader = ParquetRecordBatchReaderBuilder::try_new(file)?
            .with_row_groups(vec![rg_idx])
            .with_projection(mask.clone())
            .build()?;

        // Read, evaluate, count
        let batch = reader.collect_batch();
        let match_count = eval_pattern_in_batch(&batch, &predicates);
        match_count
    })
    .collect();

let total = counts.iter().sum::<usize>() + boundary_matches;

Each task opens its own file handle, which avoids lock contention on shared file descriptors. Rayon's work-stealing scheduler keeps load balanced across cores.

Row group data is freed immediately after processing, which eliminates the 1.4-second deallocation overhead you get from holding all data until the end.

3.4 Cross-row-group boundary handling

Pattern matches can span row group boundaries (the first half of a pattern at the end of RG1, the second half at the start of RG2). RgBoundaryInfo stores boundary rows from each RG:

rust

struct RgBoundaryInfo {
    first_rows: Vec<RecordBatch>,  // First N rows of each RG
    last_rows: Vec<RecordBatch>,   // Last N rows of each RG
}

After parallel processing of individual RGs, a sequential scan processes boundary regions for cross-RG matches.

3.5 Schema unification

Different row groups in a Parquet file may have different column types (e.g., Utf8View vs Utf8). parallel_scan.rs performs schema unification before merging results:

rust

// Handle Utf8View -> Utf8 conversion
if col.data_type() == &DataType::Utf8View {
    let view_arr = col.as_any().downcast_ref::<StringViewArray>()?;
    let string_arr: StringArray = view_arr.iter().collect();
    // Replace with Utf8 type
}

4. Vectorized pattern matching (`src/ops/pattern_match.rs`, 1,091 lines)

4.1 Problem definition

Given N step predicates and an optional partition key, find all positions where consecutive rows within the same partition match each step predicate in sequence. The canonical use case: funnel analysis.

python

# Three-step funnel: homepage -> product page -> checkout
events.search_pattern_count(
    lambda r: r.url.s.starts_with("/"),
    lambda r: r.url.s.starts_with("/product"),
    lambda r: r.url.s.starts_with("/checkout"),
    partition_by="userid"
)

4.2 Algorithm design

The traditional approach uses LEAD() window functions to compute the next N-1 rows' values for every row, then filters the entire table. This requires N-1 full-table computations.

LTSeq's algorithm uses sparse evaluation (as documented in pattern_match.rs):

Step 1: Evaluate step1 predicate for ALL rows -> BooleanArray     (full scan)
Step 2: Only where step1=true, check if row[i+1] matches step2    (sparse, ~0.1%)
Step 3: Only where step1+step2 both match, check row[i+2]         (even sparser)

Since step1's match rate is typically only ~0.1%, subsequent steps require negligible computation.

4.3 The eval_predicate function

The core predicate evaluator (pattern_match.rs:44) recursively evaluates an expression tree against a RecordBatch:

rust

pub(crate) fn eval_predicate(
    expr: &PyExpr,
    batch: &RecordBatch,
    name_to_idx: &HashMap<String, usize>,
) -> Result<BooleanArray, String> {
    let arr = eval_expr(expr, batch, name_to_idx)?;
    arr.as_any()
        .downcast_ref::<BooleanArray>()
        .cloned()
        .ok_or_else(|| format!("Predicate must evaluate to boolean, got {:?}", arr.data_type()))
}

It supports column references, literals, binary operations, unary operations, and string methods (starts_with, ends_with, contains), which are the common building blocks of funnel predicates.

4.4 Partition boundary detection via prefix sum

How do you efficiently determine whether row[i] and row[i+k] are in the same partition? LTSeq uses a prefix sum to reduce O(k) comparisons to O(1):

rust

// Build partition ID array: same partition key as previous row -> same ID, different -> increment
let partition_ids: Vec<u64> = ...;  // prefix sum

// Check: pids[i] == pids[i + num_steps - 1]
// means rows [i, i+num_steps-1] are all in the same partition
if partition_ids[i] == partition_ids[i + num_steps - 1] {
    // Same partition, proceed to check step matches
}

This is a classic algorithmic optimization: precompute O(N) once, then answer O(1) per query.

4.5 String prefix optimization

extract_starts_with_prefix() detects whether a predicate is of the form url.starts_with("/product"). If so, it can use Arrow's string comparison directly instead of the regular expression engine:

rust

pub fn extract_starts_with_prefix(expr: &PyExpr) -> Option<(String, String)> {
    match expr {
        PyExpr::Call { func, args, on, .. } if func == "str_starts_with" => {
            if let (PyExpr::Column(col_name), Some(PyExpr::Literal { value, .. })) =
                   (&**on, args.first()) {
                Some((col_name.clone(), value.clone()))
            } else { None }
        }
        _ => None,
    }
}

This optimization is particularly effective for ClickBench's URL funnel queries, where all step predicates are starts_with checks.

4.6 Count-only path

When only the count is needed (not the matching rows), search_pattern_count_impl avoids building the full result table:

rust

// Instead of:
// 1. Find all matching row indices
// 2. Build result RecordBatch with take()
// 3. Wrap in LTSeqTable
// 4. Call count()

// Just:
// 1. Count matching positions
// 2. Return integer

This saves the cost of arrow::compute::take() and RecordBatch construction for potentially hundreds of thousands of matches.

5. As-of join: binary search for time-series alignment (`src/ops/asof_join.rs`)

As-Of Join is a fundamental time-series operation: match each row with the "nearest" row from another table.

5.1 Three directions

rust

/// Find largest index where right_times[idx] <= target ("most recent historical data")
fn find_asof_backward(target: i64, right_times: &[i64]) -> Option<usize> {
    let idx = right_times.partition_point(|&t| t <= target);
    if idx == 0 { None } else { Some(idx - 1) }
}

/// Find smallest index where right_times[idx] >= target ("nearest future data")
fn find_asof_forward(target: i64, right_times: &[i64]) -> Option<usize> {
    let idx = right_times.partition_point(|&t| t < target);
    if idx >= right_times.len() { None } else { Some(idx) }
}

/// Find nearest index (backward bias on ties)
fn find_asof_nearest(target: i64, right_times: &[i64]) -> Option<usize> {
    let backward = find_asof_backward(target, right_times);
    let forward = find_asof_forward(target, right_times);
    match (backward, forward) {
        (Some(b), Some(f)) => {
            let diff_back = target - right_times[b];
            let diff_fwd = right_times[f] - target;
            if diff_back <= diff_fwd { Some(b) } else { Some(f) }
        }
        (b, f) => b.or(f),
    }
}

partition_point is Rust's standard library binary search, at O(log M). Total complexity is O(N log M), where N is left rows and M is right rows.

5.2 Result assembly with Arrow take()

After matching is complete, Arrow's take() operation performs efficient index-based extraction:

rust

let indices_array = UInt32Array::from(matched_right_indices);  // [Some(5), None, Some(3), ...]
for col_idx in 0..right_schema.fields().len() {
    let col = right_batch.column(col_idx);
    let result = arrow::compute::take(col, &indices_array, None)?;
    result_columns.push(result);
}

take() is Arrow's gather operation, which uses SIMD acceleration under the hood. None indices automatically generate NULL values, handling unmatched rows without special cases.

6. Execution path selection

LTSeq's group_ordered has three execution paths, tried in priority order:

                      +-- Has Parquet path?
                      |   Yes -> direct_streaming_group_ordered (fastest)
                      |          Bypasses DataFusion, reads Parquet directly
group_ordered(cond) --+
                      |   +-- Expression contains shift(1) and is supported?
                      +-- |   Yes -> linear_scan (fast)
                      |   |          Single-pass fused evaluation
                      |   |
                      +---+-- No  -> DataFusion path (general)
                                     LAG + IS DISTINCT FROM + SUM

search_pattern follows a similar pattern:

                          +-- Has Parquet path + predicates are directly evaluable?
search_pattern_count -----+   Yes -> parallel_pattern_match_count (fastest)
                          |          Rayon parallel over row groups
                          |
                          +-- No  -> DataFusion path
                                     collect + eval_predicate + single-pass scan

The selection is automatic and transparent to the user. The API is identical regardless of which path is taken.

7. Summary: specialized algorithms vs. general engines

Three things come out of this.

General engines are not universal. DataFusion is an excellent SQL engine, but its query planner doesn't understand ordered sequence semantics. For sequential operations, specialized algorithms get 1.4x to 1.5x better performance.

Fusion is where most of the gain comes from. Merging multiple full-table scans into a single pass and eliminating intermediate array allocations is the largest single source of speedup. Three scans versus one isn't a 3x difference; factor in memory allocation, cache pressure, and GC overhead and it's more than that.

Constant factors matter too, not only complexity class. TypedColumn's compile-time type elimination, timestamp zero-copy reinterpretation, and prefix-sum partition boundary detection are all constant-factor optimizations, and they add up. At 100M rows, saving 1 nanosecond per row saves 100 milliseconds.

Next article: ClickBench's 100M-row dataset, used to quantify the actual gains from these optimizations, with head-to-head benchmarks against DuckDB.

English Series

中文系列

The 2,400-Line Bytecode Engine: Inside LTSeq's Core Algorithms

1. Background: the DataFusion bottleneck

2. The linear scan engine (`src/ops/linear_scan.rs`, 2,409 lines)

2.1 Eligibility check: who gets the fast path?

2.2 Runtime value system

2.3 TypedColumn: compile-time type elimination

2.4 Fused streaming evaluation

2.5 Cross-batch state: StreamState

2.6 NULL semantics: conservative by design

3. Direct Parquet parallel engine (`src/ops/parallel_scan.rs`, 984 lines)

3.1 Why bypass DataFusion?

3.2 Strategy one: sequential streaming (for group_ordered)

3.3 Strategy two: Rayon parallel (for pattern matching)

3.4 Cross-row-group boundary handling

3.5 Schema unification

4. Vectorized pattern matching (`src/ops/pattern_match.rs`, 1,091 lines)

4.1 Problem definition

4.2 Algorithm design

4.3 The eval_predicate function

4.4 Partition boundary detection via prefix sum

4.5 String prefix optimization

4.6 Count-only path

5. As-of join: binary search for time-series alignment (`src/ops/asof_join.rs`)

5.1 Three directions

5.2 Result assembly with Arrow take()

6. Execution path selection

7. Summary: specialized algorithms vs. general engines

The 2,400-Line Bytecode Engine: Inside LTSeq's Core Algorithms ​

1. Background: the DataFusion bottleneck ​

2. The linear scan engine (src/ops/linear_scan.rs, 2,409 lines) ​

2.1 Eligibility check: who gets the fast path? ​

2.2 Runtime value system ​

2.3 TypedColumn: compile-time type elimination ​

2.4 Fused streaming evaluation ​

2.5 Cross-batch state: StreamState ​

2.6 NULL semantics: conservative by design ​

3. Direct Parquet parallel engine (src/ops/parallel_scan.rs, 984 lines) ​

3.1 Why bypass DataFusion? ​

3.2 Strategy one: sequential streaming (for group_ordered) ​

3.3 Strategy two: Rayon parallel (for pattern matching) ​

3.4 Cross-row-group boundary handling ​

3.5 Schema unification ​

4. Vectorized pattern matching (src/ops/pattern_match.rs, 1,091 lines) ​

4.1 Problem definition ​

4.2 Algorithm design ​

4.3 The eval_predicate function ​

4.4 Partition boundary detection via prefix sum ​

4.5 String prefix optimization ​

4.6 Count-only path ​

5. As-of join: binary search for time-series alignment (src/ops/asof_join.rs) ​

5.1 Three directions ​

5.2 Result assembly with Arrow take() ​

6. Execution path selection ​

7. Summary: specialized algorithms vs. general engines ​

The 2,400-Line Bytecode Engine: Inside LTSeq's Core Algorithms

1. Background: the DataFusion bottleneck

2. The linear scan engine (`src/ops/linear_scan.rs`, 2,409 lines)

2.1 Eligibility check: who gets the fast path?

2.2 Runtime value system

2.3 TypedColumn: compile-time type elimination

2.4 Fused streaming evaluation

2.5 Cross-batch state: StreamState

2.6 NULL semantics: conservative by design

3. Direct Parquet parallel engine (`src/ops/parallel_scan.rs`, 984 lines)

3.1 Why bypass DataFusion?

3.2 Strategy one: sequential streaming (for group_ordered)

3.3 Strategy two: Rayon parallel (for pattern matching)

3.4 Cross-row-group boundary handling

3.5 Schema unification

4. Vectorized pattern matching (`src/ops/pattern_match.rs`, 1,091 lines)

4.1 Problem definition

4.2 Algorithm design

4.3 The eval_predicate function

4.4 Partition boundary detection via prefix sum

4.5 String prefix optimization

4.6 Count-only path

5. As-of join: binary search for time-series alignment (`src/ops/asof_join.rs`)

5.1 Three directions

5.2 Result assembly with Arrow take()

6. Execution path selection

7. Summary: specialized algorithms vs. general engines