Materialize: Support setting custom timestamp offsets for Kafka based sources

[maqroll]

4948 introduces custom start offsets for Kafka based sources in Materialize. Kafka also allows to start consuming from custom timestamp.

I have a use case that will benefit from that feature: aggregate visits topic to get most-seen-article in last N minutes (sliding window).
I keep more than N minutes depth in Kafka topic so before #4948 I have to wait until Materialize read all topic messages (more than twenty minutes).
Since #4948 (I didn't tested yet) I could query timestamp index in Kafka and set up stream using those offsets.
But would be more convenient if I could setup start offsets as timestamp or even better relative timestamp (i.e. negative values could represent now() - timestamp) directly in Materialize.

In librdkafka offsetsForTime allows to query timestamp index.

Any temporal aggregation could benefit from this feature.

Would be feasible to implement this in Materialize?

[cirego]

Hey @maqroll, I've started looking at what it would take to implement this feature and would like to double check something with you, if that's alright. The Kafka API assumes that you want a single timestamp across all partitions and I gather this is the behavior you want as well. Would something like the following work for you (where 2 is your desired timestamp)?

CREATE SOURCE KAFKA source_from_times
FROM KAFKA BROKER ... TOPIC ...
WITH (offsets_for_times = 2)
FORMAT ...

Note that this behavior is different from our start_offsets behavior, which enables setting a start offset on a per-partition basis:

CREATE SOURCE KAFKA source_from_offsets
FROM KAFKA BROKER ... TOPIC ...
WITH (start_offsets = [1,2,1,...])
FORMAT ...

[maqroll]

Hi @cirego, thanks for the quick response.

In my use case offsets_for_times = 2 (one timestamp for all the partitions) is fine.

What's your opinion about relative timestamps? I proposed using negative timestamps as relative ones (offsets_for_times = 300000, for the last five minutes).

In my use case (sliding windows) relative timestamps would be helpful because they would be valid across restarts.

[cirego]

Hi @cirego, thanks for the quick response.

In my use case offsets_for_times = 2 (one timestamp for all the partitions) is fine.

What's your opinion about relative timestamps? I proposed using negative timestamps as relative ones (offsets_for_times = 300000, for the last five minutes).

In my use case (sliding windows) relative timestamps would be helpful because they would be valid across restarts.

Hi @maqroll, given the use case, I do think that using a relative timestamp makes more sense than an absolute timestamp. And, as you mentioned, it would avoid the need to recreate your sources and views. I'll discuss this with the rest of the team and see what they think!

[awang]

Feature spec (internal): https://docs.google.com/document/d/1c3YMuX78BYS4ufYGllNjttANKvvmbNFWqunBAe9yK74/edit#heading=h.s7512jyfvu1k

[uce]

@awang @nmeagan11 I've assigned this to myself as it seems like a good issue to get more into sources.

[uce]

Fixed in https://github.com/MaterializeInc/materialize/pull/6941.