Support efficient last N rows for stream table

[devinrsmith]

We don't currently have efficient support for retaining the last N rows of a stream table.

stream.tail(N) itself is a streaming table (ie, it doesn't retain)

stream.tailBy(N) might have the semantics we are looking for, but the current implementation doesn't work for streaming tables (implemented using groupBy which doesn't work on streaming tables). .tailBy(N) may be more powerful than we need though, since it also supports grouping columns. (Follow-up question if .tailBy(N, ...) should semantically ungroup automatically or not.) That said, retaining the last N rows of a stream table per group may be desired functionality as well.

It should be relatively straightforward in either case; but the case without the grouping columns will probably be easier.

[devinrsmith]

There is a hacky workaround if you have a stream table with an increasing sequence number (like a kafka table):

last_n = kafka_stream.updateView("X=KafkaOffset % N").lastBy("X")

or

last_n = kafka_stream.updateView("X=KafkaOffset % N").lastBy("X").sort("KafkaOffset")

[devinrsmith]

Assuming a single partition - otherwise, may also want a lastBy("KafkaPartition", "X") (but then, it's not strictly last "N").