Potential bug fix for supporting sorted-indexes for partial-upsert

[tibrewalpratik17]

label: bugfix upsert

Potential fix for #12397

As called out in the issue:

With Partial Upserts, say we had 4 events for a given primary-key in the Mutable Segment: R0, R1, R2, R3. Let's also assume that the comparison column value of these events is: R0 < R1 < (R2 = R3).

If after applying the sorted column, their order changes to: R0, R1, R3, R2, then the UMM will start pointing to R2 as the valid doc.

Now based on the assumption that R3's docId will be greater than R2's docId, raising this fix as an echancement to fix in #12395 where in case of same comparison column value we also check the docID value of the old mutable segment to know for a particular record which one to keep.

We keep the max-doc-id of older segment in making this decision. Say, in the example above; we will ensure to dedup R3 as the only record.

Tested in our local cluster by adding sorted-index column to a table with constant comparison column. Ensured that we are deduping the correct record and persisting it. Added UTs for updating the resolvingComparisonTies method.

cc @ankitsultana @Jackie-Jiang

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 64.70588% with 6 lines in your changes are missing coverage. Please review.

Project coverage is 62.17%. Comparing base (59551e4) to head (f0d82d2). Report is 444 commits behind head on master.

Files	Patch %	Lines
...cal/upsert/BasePartitionUpsertMetadataManager.java	50.00%	0 Missing and 3 partials :warning:
...local/indexsegment/mutable/MutableSegmentImpl.java	85.71%	1 Missing :warning:
...t/ConcurrentMapPartitionUpsertMetadataManager.java	66.66%	1 Missing :warning:
...ava/org/apache/pinot/segment/spi/IndexSegment.java	0.00%	1 Missing :warning:

Additional details and impacted files

```diff @@ Coverage Diff @@ ## master #12544 +/- ## ============================================ + Coverage 61.75% 62.17% +0.42% + Complexity 207 198 -9 ============================================ Files 2436 2515 +79 Lines 133233 137875 +4642 Branches 20636 21341 +705 ============================================ + Hits 82274 85729 +3455 - Misses 44911 45765 +854 - Partials 6048 6381 +333 ``` | [Flag](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | |---|---|---| | [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `<0.01% <0.00%> (-0.01%)` | :arrow_down: | | [integration](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `<0.01% <0.00%> (-0.01%)` | :arrow_down: | | [integration1](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `<0.01% <0.00%> (-0.01%)` | :arrow_down: | | [integration2](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (ø)` | | | [java-11](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.14% <64.70%> (+0.43%)` | :arrow_up: | | [java-21](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.06% <64.70%> (+0.44%)` | :arrow_up: | | [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.15% <64.70%> (+0.40%)` | :arrow_up: | | [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.04% <64.70%> (+34.32%)` | :arrow_up: | | [temurin](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.17% <64.70%> (+0.42%)` | :arrow_up: | | [unittests](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.17% <64.70%> (+0.42%)` | :arrow_up: | | [unittests1](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `46.74% <0.00%> (-0.15%)` | :arrow_down: | | [unittests2](https://app.codecov.io/gh/apache/pinot/pull/12544/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `27.80% <64.70%> (+0.07%)` | :arrow_up: | Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.

[tibrewalpratik17]

One potential solution would be to persist the original doc id into a column, then use that to break tie

Yes that's what I suggested in one of the threads above. Having a virtual column like $originalDocID would also make sense and especially for upserts we can refer that to resolve comparison ties. For now, i can think of only this as a possible solution or to persist the original docID in segment-metadata somewhere but that doesn't seem optimal. Maintaining a virtual column still makes more sense. cc @klsince what are your thoughts on this?

[ankitsultana]

Can we consider adding the stream offset as a virtual column? Then we could allow users to set that virtual column as the comparison column.

[tibrewalpratik17]

Can we consider adding the stream offset as a virtual column? Then we could allow users to set that virtual column as the comparison column.

Setting offset field as a comparison column might be another discussion we are getting into. But we can use offsets virtual column to break ties in general (very similar logic to originalDocID field). Other than higher memory / disk usage compared to originalDocID field i don't see any other challenges. But yeah having offset as a column also comes with pros like easier debugging and observability around data.

[klsince]

to my best understanding on this issue, I'd +1 to keep original docId in a virtualColumn to help break tie.

for the idea of using stream offset, I'd prefer to put stream offset in a real column, assuming we may create some ingestion transformation to extract the msg offset and put into the column. Then we can use this column as one of the comparison columns, to prevent comparison ties from happening.