Support point-in-time-join on a given set of dataframes

[jhasm-ck]

Is your feature request related to a problem? Please describe.

What is a Point-in-Time (PiT) Correct Join?
A point-in-time correct join is a database operation that performs a join between two tables in a way that ensures the results reflect the state of the tables at a specific point in time.

When is a Point-in-Time (PiT) Correct Join needed when creating training data?
When constructing a snapshot of data (e.g., training data or batch inference data) from precomputed features spread across different feature groups, we often need to construct a snapshot of feature values at a specific point in time. For example, a training dataset for supervised ML is a snapshot of feature values at the time of the observation of each label in each row in the training dataset.

A problem with creating a point-in-time-correct training data snapshot is that the underlying tables (feature groups) are typically updated at different cadences by different data pipelines. As such, it is not always possible to utilize an exact time-based join to obtain the desired result. The solution is a Point-in-Time correct Join that starts with the timestamps for the labels and retrieves the most recent feature values for the features from all the tables joined with the table containing the labels.

Describe the solution you'd like
I want a solution similar to this one: https://www.hopsworks.ai/post/a-spark-join-operator-for-point-in-time-correct-joins
Pandas has merge_asof function that performs similar operation, but it is not distributed.

Describe alternatives you've considered
There is no practical alternative to this one for datasets larger than single node. For smaller datasets pandas.merge_asof can be used.

Additional context
This feature is critical for ML feature engineering, because feature values are stored with timestamps, making all features a timeseries. This might be the biggest feature blocking people from moving from Spark to Daft+Ray for feature engineering.

[xcharleslin]

Hey @jhasm-ck , providing a quick update here. We've scoped out the work necessary to support this:

1. Implement point-in-time join at the single partition level as a new join method here. The signature would be something like:

pub fn pit_join(&self, right: &Self, left_asof: &[Expr], right_times: &[Expr]) -> DaftResult<Self>

where the left table contains the "as-of" times (in left_asof column(s)) and the right table contains the join candidates (in right_times)

2. Implement the distributed pit join as a modified sort-merge join. The stages here would be:
   a) Range partition the left tables. The code for this would be similar to this.
   b) Range partition the right tables. There are a couple of options here: either follow the same scheme as the left tables (allowing for 1-to-1 joins in the join stage but risking data skew), or partition by its own boundaries to reduce data skew, but requiring overlapping 1-to-m joins in the join stage.
   c) Execute the join. For 1-to-1 joins, the existing join logic can probably be largely reused; for 1-to-m joins, new logic would have to be written.

3. Implement a pit join interface at the API level and at the query plan level, and hook it up to the physical execution from step (2).

We anticipate that this, once begun, will take 2-4 weeks to be released inclusive of implementation, testing, and release cycle.