Preserve Kafka Message Metadata in Pinot Tables

[mapshen]

Opening this issue as per the conversation in Slack.

When streaming from Kafka, Pinot currently lacks of a way to allow users to uniquely identify messages. We may achieve it by exposing the metadata of Kafka messages such as offset/timestamp.

The approach suggested by @kishoreg is

What we need is Kafka decoder to read this metadata and add it to generic row.

CC: @mayankshriv @npawar @xiangfu0

[mapshen]

@kishoreg @npawar Supposedly the offset can be added to a decoded row by modifying [1], now I have a few questions regarding the implementation details:

• What would a good column name to follow our convention? $offset or offset?
• Should we make it configurable or always present like the virtual columns?
• Where should we define its data type when it is indexed [2]? We do this for other columns with a schema.
• What index should we use for it? Raw value or sorted forward index?

[1] https://github.com/apache/incubator-pinot/blob/master/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaJSONMessageDecoder.java#L62
[2]https://github.com/apache/incubator-pinot/blob/master/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java#L516

[npawar]

i like the idea of making it a virtual column, and having it be always present, “$offset”

since we wont know the datatype of the offset (kafka is long, kinesis sequence numbers exceed long range, hence are string) we could keep the offset datatype as string?
this would have serialized value of whatever the stream’s implementation of StreamPartitionMsgOffset is

a segment can have only 1 sorted index. So it wont be a good idea to make it have sorted index. Do we need index by default?

[mcvsubbu]

+1 to keeping it virtual column
+1 to making it configurable, since it can be quite some overhead
+1 to keeping it a string, pinot is transparent to the stream underneath.
We should not be building indices on it. Please use raw index. It is supported for consuming segments now.

In case of kinesis, we should (may want to) also keep track of other metadata like partition IDs in the group during the time the segment was being consumed. I think these do not change (if they do, we close the segment), but @npawar or @KKcorps can comment on that.

Since this is stream dependent, I would make it a string that has (at the minimum) the StreamMsgOffset serialized, and also the partition group ID. Beyond that, each stream may add its own stuff.

Also, consider having a less verbose version of this by having some data common to the entire segment, in the segment metadata (Some of these are there in zk metadata). For kafka, this could mean start/end offset, partition group id, etc.

[mcvsubbu]

@mapshen curious what your use case is for this feature. It adds significant overhead to realtime consuming memory. When we do have an error, we put out the offset that caused the error. I don't see the advantage of having the offset information for each row.

[suddendust]

If this feature is a go, I can pick it up (it might take me a while to implement it tho as I am still navigating my way around Pinot. So hopefully it's not an urgent feature).

[mcvsubbu]

@mapshen has not described a use case for this yet. For reasons that I mentioned before, I am reluctant to include this a a feature

[mapshen]

@mcvsubbu Okay, let me formalize the use case here:

As a user, I would like to have a way to uniquely identify a row/message between two queries. It would be really helpful when:

• I don't want to reprocess, or even fetch, the rows I have already seen
• I don't want to miss any messages/rows in between since the last fetch

We talked over Slack about it if you recall :)

@suddendust, appreciate your offering help. My impression was actually that we agreed to add this in principle but I just couldn't find time to write up a proposal on how to approach this. Perhaps you can start with a proposal?

[mcvsubbu]

Sorry, there was no discussion in the slack channel after I got added to it. I saw the mention for a meeting and I was hoping that one would be called, but I did not get any invite. None of these requirements were mentioned in the slack channel.

What are your queries ? (are they like select * and you want to know which new rows came in since you executed the same query previous time?

Pinot guarantees exactly once consumption, and guarantees not missing any row from underlying stream (unless underlying stream has a low retention). Since I see the slack channel has a lot of discussion about actual implementation, perhaps you already know about these guarantees, but I thought i will mention anyway

[sajjad-moradi]

@mapshen Can you elaborate a bit on why this use case is needed? Maybe it makes it easy to understand, as Subbu mentioned, if you provide queries for this use case!

[mapshen]

Sure. Say we have an inventory table of products ingested from a single-partition topic, and, to begin with, a query like

select count, offset from inventory where product = 'A' order by offset desc limit 1

is executed, from which I record the offset.

Next time a query is issued, if I don't want to reprocess, or even fetch, the rows I have already seen, and only care about the latest change if any, I would do

select count, offset  from inventory where offset > <last_recorded_offset> and product = 'A' order by offset desc limit 1

If I don't want to miss any messages/rows in between since the last fetch, I could do

select count, offset from inventory where offset > <last_recorded_offset> and product = 'A' 

From here, I would record the new offset again (if any) and repeat.

There was more context in the original thread on Slack, but I couldn't find it anymore. Seems all messages prior to 06/16 got purged (?).

[mcvsubbu]

Can you use the UPSERT feature, which updates the rows that have come in?

[mapshen]

Unfortunately no. We don't always only care about the latest.

Also, this UPSERT feature has its limitations as documented here [1], of which the required partition expansion is a showstopper already.

On top of it, we need to break ties correctly and reliably when it happens, for which we cannot rely on the time column, as there might be 2 messages with the same timestamp and primary key but different offsets.

[1] https://docs.google.com/document/d/1qljEMndPMxbbKtjlVn9mn2toz7Qrk0TGQsHLfI--7h8/edit#bookmark=kix.zg5cmgwqyx7e

[kishoreg]

@mcvsubbu , this is a very powerful feature. This is equivalent to cursor API in databases. For e.g. we can use this in ThirdEye to analyze new events that come in since the last time we ran anomaly detection.

I am +1000 on this feature. Another side benefit of this is in debugging kafka lag. Kafka provides an offset API and we can compare the two to compute the lag.

btw, I am fine with this being an actual column instead of virtual. we can solve this by having a udf pull the value from header.

[yupeng9]

+1

Other systems like Presto support these kind of features. https://prestodb.io/docs/current/connector/kafka-tutorial.html

e.g.

All columns with the underscore prefix are ingested by default. I also suggest we add partition id and (possibly key), which are useful for troubleshooting, particularly for upsert tables.