Design: Point-in-Time (AsOf) Joins

[desmondcheongzx]

Tracking issue: #1400

What is a Point-in-Time Join?

A point-in-time (asof) join matches each row in the left table to the most recent row in the right table whose timestamp is at or before the left row's timestamp, within the same entity group. The temporal equivalent of "give me the last known value as of this moment."

Left (labels/observations)          Right (feature table, updated hourly)

| entity | timestamp |              | entity | timestamp | value |
|--------|-----------|              |--------|-----------|-------|
| A      | 10:15     |              | A      | 09:00     | 1.0   |
| A      | 11:45     |              | A      | 10:00     | 2.0   |
| B      | 10:30     |              | A      | 11:00     | 3.0   |
                                    | B      | 09:00     | 5.0   |
                                    | B      | 11:00     | 6.0   |

                PiT Join (by=entity, on=timestamp)

| entity | timestamp | value |
|--------|-----------|-------|
| A      | 10:15     | 2.0   |   <- A's latest value at or before 10:15 is 10:00 -> 2.0
| A      | 11:45     | 3.0   |   <- A's latest value at or before 11:45 is 11:00 -> 3.0
| B      | 10:30     | 5.0   |   <- B's latest value at or before 10:30 is 09:00 -> 5.0

Row B @ 10:30 does not match B @ 11:00 even though that row exists - it is in the future relative to the observation. And it does not require an exact match on 09:00 either. It finds the most recent preceding value.

Motivation

ML feature engineering requires constructing training datasets by joining labels with precomputed features. Feature tables are updated at different cadences by different pipelines, so exact-timestamp joins produce mostly nulls. What you actually need is the most recent feature value at or before each observation time.

This is the standard operation in feature stores (Feast, Tecton, Hopsworks) and the biggest gap blocking users from adopting Daft for ML feature engineering workflows. Pandas has merge_asof but it's single-node. Spark has no first-class implementation (rewrites to correlated subqueries, that then decorrelates to a hash join + window function aggregate). Polars and DuckDB have dedicated operators but are single-node only. Daft can be the first distributed DataFrame engine with a native, performant asof join.

Prior Art

Engine	API	Direction	Tolerance	Nearest	Distributed
DuckDB	ASOF JOIN ... ON eq AND ineq	>=, >, <=, <	Manual (extra WHERE)	No	No (single-node)
Polars	df.join_asof(on, by, strategy)	backward / forward / nearest	Yes	Yes	No (single-node)
Pandas	pd.merge_asof(on, by, direction)	backward / forward / nearest	Yes	Yes	No (single-node)
Spark	pyspark.pandas.merge_asof() (private)	backward only	Yes	No	Rewrites to correlated subquery (slow)

DuckDB hash-partitions both sides on equality keys into buckets, sorts each bucket on the asof key, then runs a two-pointer merge that stops at the first match per left row. Also has a "loop join" fallback for tiny probe tables (< 64 rows). No distributed execution.

Polars uses a single-pass O(M+N) merge on pre-sorted data without by keys. With by keys, it hash-partitions the right side, then does per-group asof scans. Parallelized across threads.

Spark has no dedicated physical operator. The RewriteAsOfJoin optimizer rule rewrites asof joins into correlated scalar subqueries with MIN_BY aggregation - does not exploit sortedness for O(M+N) merge. Tempo (Databricks Labs) uses a union + window function workaround instead.

Proposed API

def join_asof(
    self,
    other: DataFrame,
    on: ColumnInputType | None = None,                            # asof key (same name both sides)
    left_on: ColumnInputType | None = None,                       # asof key (left, if names differ)
    right_on: ColumnInputType | None = None,                      # asof key (right, if names differ)
    by: list[ColumnInputType] | ColumnInputType | None = None,    # equality keys (same name)
    left_by: list[ColumnInputType] | ColumnInputType | None = None,
    right_by: list[ColumnInputType] | ColumnInputType | None = None,
    direction: Literal["backward", "forward", "nearest"] = "backward",
    tolerance: int | float | timedelta | None = None,
    allow_exact_matches: bool = True,
    prefix: str | None = None,
    suffix: str | None = None,
) -> DataFrame

Usage

# Simple case
df.join_asof(prices, on="timestamp", by="ticker")

# Asymmetric column names
df.join_asof(prices, left_on="event_time", right_on="price_time", by="ticker")

# Multiple equality keys, with tolerance
df.join_asof(prices, on="timestamp", by=["ticker", "exchange"],
             tolerance=timedelta(minutes=5))

Design decisions

• Separate method, not how="asof" on join(). Asof joins have fundamentally different semantics from equi-joins (inequality matching, direction, tolerance). Extending join() would force every optimizer rule that pattern-matches on Join nodes to add asof guard clauses. A separate AsofJoin logical plan node is invisible to existing optimizer passes.
• on = asof key, by = equality keys. This contradicts Daft's join() where on means equality keys. But join_asof is a different method with its own docstring, and the Polars/Pandas convention is strong enough that users from either library will map instantly. The asof key is the defining parameter of the operation - it should be the primary positional argument.
• prefix/suffix carried over from Daft's existing join() for column name conflict resolution.
• v1 ships backward only. forward, nearest, tolerance, and allow_exact_matches are followups.

Implementation Plan

Phase 1: Core kernel + native runner

New asof_join kernel at the RecordBatch level. The algorithm is a modified two-pointer merge on data sorted by (by_keys, asof_key):

for each left row:
    advance right pointer while right.entity < left.entity
    advance right pointer while right.entity == left.entity
                               AND right.timestamp <= left.timestamp
        track last valid match
    emit (left_row, best_match or null)

Structurally similar to the existing merge_inner_join() state machine in merge_join.rs - the main difference is asymmetric timestamp comparison and tracking "best match so far" instead of emitting all equality matches.

Effort estimate: <1 week

Phase 2: Distributed execution (hash partitioned)

Hash-partition both sides on the equality keys (by), then run the local asof kernel per partition. This completely avoids cross-partition lookups because all rows for a given entity are co-located. Reuses existing hash repartitioning infrastructure.

Effort estimate: ~1 week

Phase 3: Extended functionality

• forward direction (first right row where right.ts >= left.ts)
• nearest direction (closest by absolute distance)
• tolerance parameter (max allowed time delta)
• allow_exact_matches=False (strictly less/greater than)

Each is an incremental change to the kernel.

Effort estimate: <1 week

Phase 4: Distributed execution (range partitioned)

For the no-equality-keys case or extreme entity skew, range partition by (by_keys, asof_key) with overlapping right partitions at boundaries. The key insight: we only need a 1-row overlap per partition boundary (prepend the last right row from partition i-1 to partition i) to produce correct results. This is strictly better than Tempo's approach, which duplicates an entire fraction of the data.

Effort estimate: 1-2 weeks

Follow-ups (not scoped)

• Join reordering: Make the optimizer's join reordering pass aware of AsofJoin nodes. Individual asof joins cannot swap left/right, but independent asof joins can be reordered by cardinality.
• SQL syntax: ASOF JOIN ... ON ... AND ... in Daft SQL.

[desmondcheongzx]

@universalmind303 raised a good follow-up offline: a top-level daft.align_timeseries() function that chains multiple asof joins as syntactic sugar.

# Instead of:
df = df1.join_asof(df2, on="timestamp").join_asof(df3, on=...).join_asof(df4, ...)

# This:
df = daft.align_timeseries(
    df1,
    df2,
    df3,
    on="timestamp",
    *other_dfs
)

Common pattern in feature engineering when aligning N feature tables to a single label table. Implementation would just fold over join_asof calls. Adding to the follow-ups list.

[paa1750]

Hey @desmondcheongzx, I was interested in #2911 and wanted asof and inequality join support in daft. I started working on both of them and have working prototypes. Would it be possible for me to contribute?

[desmondcheongzx]

Hi @paa1750, yes we're accepting contributions, please open up separate PRs for them!