fix mongo KV range scans #343
Conversation
| @@ -185,7 +186,8 @@ public KeyValueIterator<Bytes, byte[]> range( | |||
| public KeyValueIterator<Bytes, byte[]> all(final int kafkaPartition, final long minValidTs) { | |||
| final FindIterable<KVDoc> result = docs.find(Filters.and( | |||
| Filters.not(Filters.exists(KVDoc.TOMBSTONE_TS)), | |||
| Filters.gte(KVDoc.TIMESTAMP, minValidTs) | |||
| Filters.gte(KVDoc.TIMESTAMP, minValidTs), | |||
| Filters.eq(KVDoc.KAFKA_PARTITION, kafkaPartition) | |||
There was a problem hiding this comment.
we should consider adding a configuration that will create an index in MongoDB on the kafka partition
There was a problem hiding this comment.
well we're going to have to get rid of this if we want to support partition scaling. That's why I went with the approach (long ago) of using the partitioner to figure out the kafka partition based on the key
Unfortunately I never found the time to finish that PR, so it's on me
Well technically the PR was finished but you requested a change that was completely reasonable and that's when I had to pivot to whatever it was that was more important -- point being I can dig that old PR up and finally finish it if we want to do this and it needs to happen ASAP. Thoughts?
There was a problem hiding this comment.
An aside: how much work is left on that PR @ableegoldman ?
range scans on KV stores didn't consider kafka partition - this patch fixes it by adding the kafka partition into the data model. while I don't love this since it'll make it more difficult to repartition a kafka topic, it will only be the case for topologies that use range or all on KV stores (something the DSL does not leverage, so I consider it an advanced use case)