Track segments consistently for consistent upsert view

[klsince]

Recently, we added support for consistent upsert view by synchronizing between updates on segments' validDocIds bitmaps and queries reading those bitmaps in PR #12976. But from recent tests, we find that's not enough. Other than that, we also need to track segments belonging to a table partition completely and consistently.

The current segment tracking logic for upsert view is broken in those places:

1. (completeness) using _trackedSegments is not right to ensure the consistent upsert view, because this Set is updated after the segment is registered to table manager. So query can see a segment before it is included in the upsert view. If so, today, the query falls back to get segment's bitmap outside the locking logic used to ensure consistency;
2. (completeness) when committing mutable segment, a new immutable segment is created to replace the mutable one, but the mutable segment is kept intact during segment replacement for query to access. The query can't access the new immutable segment as it's not registered until replacement is done. However, the upsert metadata gets updated (mainly the map from PK to record location) during segment replacement, so the newly ingested records start to invalidate docs in the new immutable segments, instead of the old mutable segment. This causes queries to see more than expected valid docs.
3. (completeness) the server can start a consuming segment before the broker can add it to routing table, even with this early fix (PR # 11978). Because handling the change of IdealState on server and broker is not sync'ed and we don't want to sync brokers and servers anyway due to the cost and complexity.
4. (consistency) when executing a query, the server acquires segments firstly, then get their validDocIds bitmaps. But between the two steps, new segments can get added, which can be a new consuming segment or a new immutable segment used to replace the mutable segment. And the query can't access the new segments' bitmap as not acquired them, thus getting less than expected valid docs.

To address those issues

1. A new Set is used to track segments completely and consistently. A segment is always added to this Set before registering to table mgr. And server locks the Set when acquiring segments and getting their validDocIds bitmaps to ensure no segment membership changes. All upsert view related logic is moved to a new util class UpsertViewManager for easier maintenance.
2. When committing a mutable segment, we create a new segment data mgr, called DuoSegmentDataManager, to register both the mutable and immutable segment to table mgr before segment replacement starts, so that query can acquire both. Meanwhile, we update mutable segment in place during replacement, instead of keeping it intact. In this way, we don't need to block the data ingestion to provide queries a complete data view when committing a segment.
3. A query always tries to acquire the latest consuming segment, if it's not included in the broker's query request, so that the query doesn't miss the newly ingested docs, in case broker hasn't updated its routing table yet.

A few other misc fixes

1. In mutable segment, we should update _numDocsIndexed before updating the upsert metadata, otherwise, query might see one less valid docs.
2. Return an empty bitmap instead of null when if a segment doesn't have bitmap yet, so that we don't fall back to get the bitmap again out side the locking logic, as when getting again, the segment might have created its bitmap.

All the new changes in this PR are supposed to be enabled via the new feature flag upsertConfig.consistencyMode added in PR #12976, so upsert tables not using this feature shouldn't be affected.

As to tests:

1. Added more unit tests
2. And stress tested both modes with a load generator in staging env, that kept ingesting records with a known set of PKs and committing segments every few seconds

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 50.27933% with 178 lines in your changes missing coverage. Please review.

Project coverage is 61.90%. Comparing base (59551e4) to head (8861396).
Report is 900 commits behind head on master.

Files	Patch %	Lines
.../pinot/segment/local/upsert/UpsertViewManager.java	59.70%	46 Missing and 8 partials ⚠️
...core/query/executor/ServerQueryExecutorV1Impl.java	31.37%	32 Missing and 3 partials ⚠️
...psert/ConcurrentMapTableUpsertMetadataManager.java	3.57%	27 Missing ⚠️
...ata/manager/realtime/RealtimeTableDataManager.java	0.00%	22 Missing ⚠️
...cal/upsert/BasePartitionUpsertMetadataManager.java	51.85%	6 Missing and 7 partials ⚠️
...local/indexsegment/mutable/MutableSegmentImpl.java	58.82%	4 Missing and 3 partials ⚠️
...pinot/core/data/manager/DuoSegmentDataManager.java	91.89%	3 Missing ⚠️
...ent/local/realtime/impl/RealtimeSegmentConfig.java	40.00%	2 Missing and 1 partial ⚠️
...t/ConcurrentMapPartitionUpsertMetadataManager.java	40.00%	1 Missing and 2 partials ⚠️
...tionUpsertMetadataManagerForConsistentDeletes.java	40.00%	1 Missing and 2 partials ⚠️
... and 4 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #13677      +/-   ##
============================================
+ Coverage     61.75%   61.90%   +0.14%     
+ Complexity      207      198       -9     
============================================
  Files          2436     2568     +132     
  Lines        133233   141926    +8693     
  Branches      20636    22080    +1444     
============================================
+ Hits          82274    87854    +5580     
- Misses        44911    47368    +2457     
- Partials       6048     6704     +656     

Flag	Coverage Δ
custom-integration1	<0.01% <0.00%> (-0.01%)	⬇️
integration	<0.01% <0.00%> (-0.01%)	⬇️
integration1	<0.01% <0.00%> (-0.01%)	⬇️
integration2	0.00% <0.00%> (ø)
java-11	61.84% <50.27%> (+0.13%)	⬆️
java-21	61.79% <50.27%> (+0.16%)	⬆️
skip-bytebuffers-false	61.86% <50.27%> (+0.12%)	⬆️
skip-bytebuffers-true	61.77% <50.27%> (+34.04%)	⬆️
temurin	61.90% <50.27%> (+0.14%)	⬆️
unittests	61.89% <50.27%> (+0.14%)	⬆️
unittests1	46.12% <14.52%> (-0.77%)	⬇️
unittests2	27.89% <35.75%> (+0.16%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.