LatestAt queries need a mechanism to provide the context of the origin time

[jleibs]

Context

LatestAt queries are very powerful, but they currently materialize data somewhat magically, without providing a mechanism to actually know where that data came from. This can be important for use-cases like:

• Doing additional queries at that specific time-point once you know "when" something happened.
  • A common pattern is for a user to log "events" when interesting things happen. It's easy to write a query that finds the last instance of an event before a particular time. But once you find the event you have no way of knowing when the event actually happened.
• Understanding how old data is relative to the query time
• Implementing interpolation functions

Example

Probably best articulated with an example.

Consider the table:

+----------+------------+-----------+
| log_tick | /my_scalar | /my_color |
+----------+------------+-----------+
|    1     |   1.2      |   RED     |
|    3     |   3.4      |    -      |
|    5     |   2.7      |   BLUE    |
|    7     |   4.1      |    -      |
+----------+------------+-----------+

When using fill_latest_at() it can be helpful to be able to answer the question: "From which time was the latest_at data filled it".

It should be possible to do something like the following:

dataset.dataframe_query_view(index="log_tick", contents="/**")
       .using_index_values([4, 10])
       .fill_latest_at(with_origin_time=True)
       .df()
       .select("log_tick",
               "/my_scalar",
               "/my_scalar:log_tick",
               "/my_color",
               "/my_color:log_tick")

Or maybe only materialize these on-demand if there was a way to articulate this as a udf?

dataset.dataframe_query_view(index="log_tick", contents="/**")
       .using_index_values([4, 10])
       .fill_latest_at()
       .df()
       .select("log_tick",
               "/my_scalar",
               udf.origin_time("/my_scalar"),
               "/my_color",
               udf.origin_time("/my_color"))

The expected output would be:

+----------+------------+---------------------+-----------+---------------------+
| log_tick | /my_scalar | /my_scalar:log_tick | /my_color | /my_color:log_tick  |
+----------+------------+---------------------+-----------+---------------------+
|    4     |   3.4      |          3          |   RED     |          1          |
|   10     |   4.1      |          7          |   BLUE    |          5          |
+----------+------------+---------------------+-----------+---------------------+

[timsaucer]

This is a common pattern and I can write up a way to do it with window functions, but those window functions can be very slow especially compared to how it would work if we did this at generation. Actually this would be a good showcase of how the platform that leans into understanding the fill forward (latest at) can outperform doing this post-hoc in dataframe operations.

[jleibs]

here's an example of a thing that's challenging:

I would like to write this as:

# Find the latest embedding for the selected time
embedding_table = (
    dataset.dataframe_query_view(index="real_time", contents="/camera/wrist/embedding").using_index_values([selected_time]).fill_latest_at().df()
    .sort("embedding:real_time")
    .limit(1)
)
embedding = embedding_table.select("/camera/wrist/embedding:embeddings").collect()[0][0][0].as_py()

But I can't sort these results because every latest_at will give me the exact same time stamp. So every partition that occurs after selected_time gives me an indistuinguishable result.

Instead I have to do a full scan of the entire dataset:

# Find the latest embedding for the selected time
embedding_table = (
    dataset.dataframe_query_view(index="real_time", contents="/camera/wrist/embedding").df()
    .filter(df.col("real_time") <= df.functions.to_timestamp_nanos(df.lit(selected_time)))
    .sort(df.expr.SortExpr(expr=df.col("real_time"), ascending=False, nulls_first=False))
    .limit(1)
)
embedding = embedding_table.select("/camera/wrist/embedding:embeddings").collect()[0][0][0].as_py()

[teh-cmc]

I've done some exploratory work of this, and the most important discovery was that this requires the introduction of new concept in sorbet. Specifically, we need to be able to express that a column of index values relates to a specific data-column descriptor (e.g. "here are nanoseconds timestamp values, associated with global index log_time, scoped to /a/b/c@Points3D:Positions3D#positions).

I'll do a proof of concept in that direction when I get a minute (actually computing the data itself doesn't seem that hard).