xiangfu0 commented on code in PR #19817:
URL: https://github.com/apache/pinot/pull/19817#discussion_r4278298688
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1765,26 +1768,40 @@ public void goOnlineFromConsuming(SegmentZKMetadata
segmentZKMetadata)
protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata)
throws Exception {
- boolean allowedDuringDownload =
_parallelSegmentConsumptionPolicy.isAllowedDuringDownload();
// A local build that failed or mismatched the committed CRC has already
released the semaphore when the policy
// allows consumption during build, so the next consuming segment can be
running during this download as well
boolean releasedDuringBuild =
_parallelSegmentConsumptionPolicy.isAllowedDuringBuild() &&
!_consumerSemaphoreAcquired.get();
- if ((allowedDuringDownload || releasedDuringBuild) &&
_partitionUpsertMetadataManager != null
- &&
_partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment))
{
- // Table config validation rejects this in PROTECTED mode. It still
happens when the mode is switched after the
- // table is created, when the server-level default allows partial upsert
consumption during commit, or when a
- // pauseless table falls back to a download after its local build fails.
- _segmentLogger.error("Next consuming segment can run during download
while this replace reverts upsert "
- + "metadata. Its snapshot can miss the rows the revert restores. Use
DISALLOW_ALWAYS");
- _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ // Hold the semaphore when the replace rewrites the key map, so the next
consuming segment does not start against
+ // a partially reverted one, and does not replay rows against keys this
replace is about to remove.
+ boolean holdsConsumerSemaphore = holdsConsumerSemaphoreForMetadata();
+ boolean allowedDuringDownload =
+ _parallelSegmentConsumptionPolicy.isAllowedDuringDownload() &&
!holdsConsumerSemaphore;
+ if (releasedDuringBuild && holdsConsumerSemaphore) {
Review Comment:
The error log and the `UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD` meter
now only fire when `releasedDuringBuild` is true. For
`ALLOW_DURING_DOWNLOAD_ONLY`, `isAllowedDuringBuild()` is false, so
`releasedDuringBuild` is structurally always false and this branch can never
run — yet that policy with an upsert revert is exactly what the old code
flagged. The new code safely holds the semaphore there instead of releasing,
which is good, but the operator loses the "your PROTECTED config is unsafe,
switch to DISALLOW_ALWAYS" signal entirely. Consider still logging/metering
when `holdsConsumerSemaphore && isAllowedDuringDownload()` so a
silently-overridden policy is still visible.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1207,7 +1207,10 @@ SegmentBuildDescriptor getSegmentBuildDescriptor() {
@Nullable
protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit)
throws SegmentBuildFailureException {
- if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()) {
+ // Not dedup: it removes a segment's keys when the mutable segment is
offloaded, which is after the replace, so
Review Comment:
Same as the `holdsConsumerSemaphoreForMetadata` doc below: "nothing it owns
is rewritten while the build runs" isn't accurate — the dedup key re-point
happens at the replace that the build leads to, on both the build and download
paths. Releasing during build is still fine (the exposed window at build-end is
tiny), but it'd help to phrase the justification as a window-size trade-off
rather than "nothing is rewritten."
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1765,26 +1768,40 @@ public void goOnlineFromConsuming(SegmentZKMetadata
segmentZKMetadata)
protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata)
throws Exception {
- boolean allowedDuringDownload =
_parallelSegmentConsumptionPolicy.isAllowedDuringDownload();
// A local build that failed or mismatched the committed CRC has already
released the semaphore when the policy
// allows consumption during build, so the next consuming segment can be
running during this download as well
boolean releasedDuringBuild =
_parallelSegmentConsumptionPolicy.isAllowedDuringBuild() &&
!_consumerSemaphoreAcquired.get();
- if ((allowedDuringDownload || releasedDuringBuild) &&
_partitionUpsertMetadataManager != null
- &&
_partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment))
{
- // Table config validation rejects this in PROTECTED mode. It still
happens when the mode is switched after the
- // table is created, when the server-level default allows partial upsert
consumption during commit, or when a
- // pauseless table falls back to a download after its local build fails.
- _segmentLogger.error("Next consuming segment can run during download
while this replace reverts upsert "
- + "metadata. Its snapshot can miss the rows the revert restores. Use
DISALLOW_ALWAYS");
- _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ // Hold the semaphore when the replace rewrites the key map, so the next
consuming segment does not start against
+ // a partially reverted one, and does not replay rows against keys this
replace is about to remove.
+ boolean holdsConsumerSemaphore = holdsConsumerSemaphoreForMetadata();
+ boolean allowedDuringDownload =
+ _parallelSegmentConsumptionPolicy.isAllowedDuringDownload() &&
!holdsConsumerSemaphore;
+ if (releasedDuringBuild && holdsConsumerSemaphore) {
+ // The build holds the semaphore for these tables, so reaching this
means it was released on another path.
+ _segmentLogger.error("Next consuming segment can run during download
while this replace rewrites upsert or "
+ + "dedup metadata. Its rows can be dropped against keys this replace
removes. Use DISALLOW_ALWAYS");
+ if (revertsUpsertMetadata()) {
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L);
+ }
}
if (allowedDuringDownload) {
closeStreamConsumerAndReleaseSemaphore();
}
_realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata);
}
+ /// The replace rewrites the key map, so the next consuming segment must not
run against it during the download:
+ /// upsert reverts those keys, dedup removes them when the mutable segment
is offloaded.
Review Comment:
This premise isn't quite right: dedup rewrites the key map at the *replace*,
not at offload. `handleDedup` -> `replaceSegment` re-points every key from the
mutable to the new immutable segment inside both `replaceConsumingSegment`
(build path) and `downloadAndReplaceConsumingSegment`; offload only removes
whatever keys remain. The held-during-download behavior is still safe, but the
stated reason is misleading — the real distinction is the length of the window
T2 is exposed, not where the rewrite happens.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1207,7 +1207,10 @@ SegmentBuildDescriptor getSegmentBuildDescriptor() {
@Nullable
protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit)
throws SegmentBuildFailureException {
- if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()) {
+ // Not dedup: it removes a segment's keys when the mutable segment is
offloaded, which is after the replace, so
+ // nothing it owns is rewritten while the build runs. An upsert revert is
rewritten at the replace the build
+ // leads to, and the next consuming segment must not already be running
against the key map by then.
+ if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild() &&
!revertsUpsertMetadata()) {
Review Comment:
This new build-side guard is the core behavior change, but no test covers it
— the test `FakeRealtimeSegmentDataManager.buildSegmentInternal` override
replaces this whole method, so the `&& !revertsUpsertMetadata()` branch is
never exercised. Combined with the still-open P2 (no
`ALLOW_DURING_DOWNLOAD_ONLY`/`ALLOW_ALWAYS` + revert download case), the
semaphore-hold behavior could regress silently. Worth a focused test asserting
the permit stays held through build/download when reverting and is released
after offload.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
