Flush Segments on a Schedule

[mapshen]

Currently, a segment can be flushed based on the size, a number of rows, or time since creation. It would be great if there is a way to trigger a flush based on a (cron) schedule.

The motivation is that it can take minutes to flush a segment, during which Pinot stops consuming new messages and hence there would be a delay of minutes. This may totally throw the users off for time-sensitive use cases.

[mayankshriv]

Are you saying that the consumption stops for minutes? If so, perhaps the existing config to control segment size are not tuned appropriately?

[Jackie-Jiang]

It will be good if we can add a controller API to flush the consuming segment. We can use this API to achieve the schedule based flush. Also, it would be useful for real-time schema evaluation, where the new columns can only be picked by next consuming segment.

[mcvsubbu]

Please don't add controller API to flush segments. It will lead to race conditions. Right now, servers drive the protocol, and let it be like that. If you desparately want to add a controller driven segment completion, then send a helix message to the participants and let them initiate the protocol, but I would like to make sure that this complexity is needed.

I agree with @mayankshriv that we need to re-look at the segment size first. Perhaps it is too big? @mapshen did you run the realtime provison helper? Can you cut/paste the output here?

[sajjad-moradi]

IIUC the problem here is during segment commit, there's no consumption. Instead of having a time scheduled for segment commit, we can find a way to start consuming for the new segment while the current segment is being committed.

[mapshen]

IIUC the problem here is during segment commit, there's no consumption. Instead of having a time scheduled for segment commit, we can find a way to start consuming for the new segment while the current segment is being committed.

@sajjad-moradi It would be great if we can start consuming without waiting for the current segment.

did you run the realtime provison helper? Can you cut/paste the output here?

@mcvsubbu I did before but didn’t find it very helpful. I can try redoing it later. That said, how fast could it be after tuning it to the bone in your experience?

Are you saying that the consumption stops for minutes?

@mayankshriv Yes, like Jackie put it earlier,

With different columns and indexes, the segment creation time could be very different. Normally it might take ~3 minutes to build the segment.

An ideal state for us would be a few seconds in the 99th percentile.

[Jackie-Jiang]

Please don't add controller API to flush segments. It will lead to race conditions. Right now, servers drive the protocol, and let it be like that. If you desparately want to add a controller driven segment completion, then send a helix message to the participants and let them initiate the protocol, but I would like to make sure that this complexity is needed.

I agree with @mayankshriv that we need to re-look at the segment size first. Perhaps it is too big? @mapshen did you run the realtime provison helper? Can you cut/paste the output here?

@mcvsubbu Yes, the commit should be driven by the server. Controller should send message to servers and ask them to start the commit process.

[mcvsubbu]

Repeating my message to @mapshen in the slack channel:
Please re-run the realtime prov helper and paste the output in this issue.

There is no standard for the time taken to build a segment. It is a function of how large your segment is (which includes number of rows, cardinality of columns, number of columns, length of string columns, the type and number of indices you have in there, etc.). In general, DO NOT use number of rows as the limit. It is better to use segment size as the limit. I hope you are doing that. If so, can you share what the size is, and can you reduce the size? It is also possible that all partitions on the machine are completing the segment at the same time. How many partitions do you have on any one machine? It is possible that your machine is (a) swapping or (b) GC-ing or both. What is your heap size? Also, what is your QPS? How much delay can you tolerate for incoming freshness of data? Do you know that Pinot provides a way to measure the freshness of data in your query response?

There are knobs that you can tune before starting to propose protocol changes (that can cause unnecessary race conditions in the code and issues for other installations).

What is your ingestion rate (per partition)? both in number of messages and number of bytes?

[mcvsubbu]

IIUC the problem here is during segment commit, there's no consumption. Instead of having a time scheduled for segment commit, we can find a way to start consuming for the new segment while the current segment is being committed.

+1 to this, if we can write up a design doc to do it, and take it up.

Here is one idea that I have toyed with (only in thought-ware):

When the current consuming segment (say, segment A) starts the completion phase, start a new consuming segment,( say A+) that is a MutableSegment as well. Whenever queries come in for segment A, include A+ as well in the server, so that the latest rows are included.

Happy path: Segment A goes to ONLINE state at some point, and continues happily ever after. Segment A+ is renamed to segment B (the next one in line) when the server gets an OFFLINE to CONSUMING state transition.

Unahappy paths are essentially unanswered.

• We now need a lot more memory on the system. May not be double (if mmaped), but some extra.
• What happens if segment A takes a long time to complete, and it is a high ingestion use case? Can we stop consumption in A+ after reaching (say) 10% of expected number of rows? Thereafter you get stale data until something happens to A.
• Other unhappy questions that I may not even know about.

[mcvsubbu]

Another thing to do in this area is to avoid multiple partitions getting into the segment completion phase at the same time. This can be done from the controller. For different partitions, it can provide a slightly increased or decreased number of rows as threshold. The variations may have to be significant for high ingestion use cases, so this will. have to be a function of ingestion rate. This will move us towards a system where at any given time, only some partitions are awaiting completion while others are consuming. Presumably a high ingestion system has several partitions so that we see new rows all the time, and for a low ingestion rate, it may not matter too much.