Early publishing segments in IndexTask

[jihoonson]

As discussed here, the current implementation of IndexTask has a potential out of disk problem, which is caused by keeping all segments in a local disk until publishing them at the very end of IndexTask.
The solution what I'm considering is publishing segments early when the size of generated segments exceeds a predefined threshold in the middle of segment generation. Thus, the behavior of FiniteAppenderatorDriver.add() will be changed to

1. Add a row to appenderator.
2. Check the size of segments generated so far.
3. If it exceeds the predefined threshold,
   3.1) Persist all data indexed through FiniteAppenderatorDriver so far.
   3.2) Publish all data indexed through FiniteAppenderatorDriver so far, and waits for segment hand-off. This is because data stored in a local disk are removed after hand-off.
4. Repeat from 1 to 3 until all data are read from the firehose.

Also, this proposal requires to use LinearShardSpec instead of HashBasedNumberedShardSpec when an interval needs to be partitioned. Currently, shardSpecs must be determined before starting ingestion. However, with this proposal, it is very difficult to determine required shardSpecs in advance. Instead, LinearShardSpec can be used for incremental sharding, i.e., creating a new shard whenever a segment is published.

This proposal changes the rollup mode of IndexTask from guaranteed rollup to non-guaranteed rollup when the rollup is enabled because two rows that should roll up might be in different publish cycles. I think it's ok for now because non-guaranteed rollup does not affect to the query result accuracy even though it can increase the space usage. However, we need to devise another solution to support guaranteed rollup in the near future.

I think this will also work well with KafkaIndexTask.

[gianm]

For (3.2) there's no need to wait for handoff (meaning historical nodes loading the data), we just need to wait for publishing.

Regular NumberedShardSpecs should be fine. They support "extending" beyond a core set, by setting the partitionNum to something higher than the corePartitions.

This writeup makes me realize that there's a couple of things we lose by publishing early:

• Guaranteed rollup
• Atomic publishing (linear shard specs, and numbered ones with partitions outside the core set, are published piecemeal rather than as an atomic unit)

The lack of atomic publishing would mean that replacing existing data will be done non-atomically, which for some users is not acceptable. So I guess while early publishing is a useful option for other users, we would really want an option for loading large datasets "all at once" too so we can get guaranteed rollup and atomic publishing. Probably through parallelization. It could be a later patch though.

[jihoonson]

@gianm thanks for your comment.

For (3.2) there's no need to wait for handoff (meaning historical nodes loading the data), we just need to wait for publishing.

Actually, IndexTask uses NoopSegmentHandoffNotifierFactory which does nothing, so I think it would be fine to wait for handoff. However, your another comment sounds good and it would be better if we can skip unnecessary call of NoopSegmentHandoffNotifierFactory.

The lack of atomic publishing would mean that replacing existing data will be done non-atomically, which for some users is not acceptable. So I guess while early publishing is a useful option for other users, we would really want an option for loading large datasets "all at once" too so we can get guaranteed rollup and atomic publishing. Probably through parallelization. It could be a later patch though.

Do you mean that it would be ok if I add a way for guaranteed rollup in a follow-up patch?

[gianm]

Actually, IndexTask uses NoopSegmentHandoffNotifierFactory which does nothing, so I think it would be fine to wait for handoff. However, your another comment sounds good and it would be better if we can skip unnecessary call of NoopSegmentHandoffNotifierFactory.

FYI we're having a discussion about that design on a thread in #4178 (comment) too; I think that PR and #4238 have a lot of similarity in goals. I'm hoping we can find a design that will be simple and effective for everyone…

Do you mean that it would be ok if I add a way for guaranteed rollup in a follow-up patch?

Yeah, that's what I mean. Assuming guaranteeing rollup also means doing atomic publishing (it probably could; they seem related to me).

[jihoonson]

Right, they would be related each other.
Ok. I'll first read #4178 and finish #4238, and then raise another pr for guaranteed rollup.