From 48a9c4fe9d119f1128f115fb9bc2cb0e74cdbc3a Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Sat, 10 Oct 2026 17:59:20 -0700 Subject: [PATCH 01/13] Exclude pauseless tables from upsert consistency during consumption A pauseless table consumes while the previous segment is still building, so its upsert metadata cannot be kept reconcilable across replicas. Exclude it from isTableTypeInconsistentDuringConsumption so the revert and the previous-location tracking it depends on are both skipped. --- .../local/upsert/BasePartitionUpsertMetadataManager.java | 4 ++-- .../apache/pinot/segment/local/upsert/UpsertContext.java | 6 +++++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java index 3a2c00aeab57..9d836e809d9b 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java @@ -729,8 +729,8 @@ public void replaceSegment(ImmutableSegment segment, @Nullable ThreadSafeMutable } /// Determines whether metadata should be reverted when inconsistencies are detected during segment replacement. - /// This is only applicable when in PROTECTED mode, the old segment is a mutable segment, - /// and the table has inconsistent state configurations. + /// This is only applicable when in PROTECTED mode, the old segment is a mutable segment, the table has inconsistent + /// state configurations. /// /// @param oldSegment the old segment being replaced /// @return true if metadata revert should be performed on inconsistency diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java index 625da707f943..d1b653bb2269 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java @@ -26,6 +26,7 @@ import javax.annotation.Nullable; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.builder.ToStringBuilder; +import org.apache.pinot.common.utils.PauselessConsumptionUtils; import org.apache.pinot.segment.local.data.manager.TableDataManager; import org.apache.pinot.spi.config.table.HashFunction; import org.apache.pinot.spi.config.table.TableConfig; @@ -201,8 +202,11 @@ public File getTableIndexDir() { /// during segment replacement after force commit. This happens when: /// - Partial upsert is enabled (records need to be merged with previous values) /// - dropOutOfOrderRecord is enabled with NONE consistency mode (records may have been dropped) + /// Pauseless tables are excluded: they consume while the previous segment is still building, so the upsert + /// metadata cannot be kept reconcilable across replicas. public boolean isTableTypeInconsistentDuringConsumption() { - return _dropOutOfOrderRecord || _outOfOrderRecordColumn != null || _partialUpsertHandlerSupplier != null; + return (_dropOutOfOrderRecord || _outOfOrderRecordColumn != null || _partialUpsertHandlerSupplier != null) + && !PauselessConsumptionUtils.isPauselessEnabled(_tableConfig); } @Override From 8f6ad653058a37029b1fc4150ccba01e0432aefd Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Sat, 10 Oct 2026 18:47:49 -0700 Subject: [PATCH 02/13] Hold the consumer semaphore when the replace reverts upsert metadata Reuse the condition the warning already computes, so the semaphore is not released before a download whose replace will revert. The warning narrows to the case the build already released it and it cannot be held back. --- .../realtime/RealtimeSegmentDataManager.java | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index c5209138ff43..fe2244731ea0 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1765,16 +1765,18 @@ 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. + boolean revertsUpsertMetadata = _partitionUpsertMetadataManager != null + && _partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment); + // Hold the semaphore when the replace reverts, so the next consuming segment does not start against a partially + // reverted primary key map. + boolean allowedDuringDownload = + _parallelSegmentConsumptionPolicy.isAllowedDuringDownload() && !revertsUpsertMetadata; + if (releasedDuringBuild && revertsUpsertMetadata) { + // The build already released the semaphore, so it cannot be held back here. _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); From 7406c742bb7e35121a25fa4ed6e57da68887090f Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Sun, 11 Oct 2026 22:30:53 -0700 Subject: [PATCH 03/13] Drop the unrelated comment rewrap from BasePartitionUpsertMetadataManager The only change this PR carried in the file was a two-line rewrap of a doc comment, which also lost the "and" joining the clauses. Nothing in the PR depends on it. Co-Authored-By: Claude Opus 5 (1M context) --- .../local/upsert/BasePartitionUpsertMetadataManager.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java index 9d836e809d9b..3a2c00aeab57 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java @@ -729,8 +729,8 @@ public void replaceSegment(ImmutableSegment segment, @Nullable ThreadSafeMutable } /// Determines whether metadata should be reverted when inconsistencies are detected during segment replacement. - /// This is only applicable when in PROTECTED mode, the old segment is a mutable segment, the table has inconsistent - /// state configurations. + /// This is only applicable when in PROTECTED mode, the old segment is a mutable segment, + /// and the table has inconsistent state configurations. /// /// @param oldSegment the old segment being replaced /// @return true if metadata revert should be performed on inconsistency From 088163fce3358a667cbc4d5df1b1921919573354 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Mon, 12 Oct 2026 13:59:53 -0700 Subject: [PATCH 04/13] Hold the consumer semaphore through the build too, and for dedup tables buildSegmentInternal() released the semaphore in its first statement whenever the policy allowed consumption during build. A build that throws, or whose local CRC does not match the committed one, then falls back to downloadSegmentAndReplace(), where the hold this PR added has nothing left to hold: the slot is already gone and the branch can only log. Gate that release on the same predicate, so these tables keep the slot across build and download the way DISALLOW_ALWAYS does. Dedup has the same race and is covered here rather than in a follow-up. A replica that consumed past the committed end offset downloads the committed copy, and offloading the mutable segment removes the keys those extra rows added. A next consuming segment running in between replays the rows, finds the keys still pointing at the mutable segment and drops them as duplicates, so they land in neither segment. The check is not keyed on allowDedupConsumptionDuringCommit: that flag is deprecated and only read when no parallelSegmentConsumptionPolicy is set, so an explicit ALLOW_ALWAYS would skip the hold it needs. The UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD meter stays upsert-only. --- .../realtime/RealtimeSegmentDataManager.java | 45 ++++++++++++++----- 1 file changed, 34 insertions(+), 11 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index fe2244731ea0..48a198d6a306 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1207,7 +1207,9 @@ SegmentBuildDescriptor getSegmentBuildDescriptor() { @Nullable protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) throws SegmentBuildFailureException { - if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()) { + // Not released when the replace rewrites upsert or dedup metadata: a build that fails or mismatches the + // committed CRC falls back to a download, which cannot hold back a semaphore already given away. + if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild() && !holdsConsumerSemaphoreForMetadata()) { closeStreamConsumerAndReleaseSemaphore(); } // Do not allow building segment when table data manager is already shut down @@ -1769,17 +1771,18 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) // allows consumption during build, so the next consuming segment can be running during this download as well boolean releasedDuringBuild = _parallelSegmentConsumptionPolicy.isAllowedDuringBuild() && !_consumerSemaphoreAcquired.get(); - boolean revertsUpsertMetadata = _partitionUpsertMetadataManager != null - && _partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment); - // Hold the semaphore when the replace reverts, so the next consuming segment does not start against a partially - // reverted primary key map. + // 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() && !revertsUpsertMetadata; - if (releasedDuringBuild && revertsUpsertMetadata) { - // The build already released the semaphore, so it cannot be held back here. - _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); + _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(); @@ -1787,6 +1790,26 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) _realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata); } + /// Whether replacing this segment rewrites the key map its consuming rows wrote, so the next consuming segment + /// must not start until the replace is done. Such tables keep the consumer semaphore across both the build and the + /// download, which is what DISALLOW_ALWAYS does. + /// + /// Upsert: the replace reverts the metadata the consuming rows wrote, and the next segment would start against a + /// partially reverted primary key map. Dedup: a replica that consumed past the committed end offset downloads the + /// committed copy, and offloading the mutable segment removes the keys those extra rows added; a next consuming + /// segment running in between replays the rows, finds the keys still pointing at the mutable segment and drops + /// them as duplicates, so they land in neither segment. Dedup is not keyed on allowDedupConsumptionDuringCommit: + /// that one is deprecated and only read when no parallelSegmentConsumptionPolicy is set, so an explicit + /// ALLOW_ALWAYS would skip the hold it needs. + private boolean holdsConsumerSemaphoreForMetadata() { + return revertsUpsertMetadata() || _partitionDedupMetadataManager != null; + } + + private boolean revertsUpsertMetadata() { + return _partitionUpsertMetadataManager != null + && _partitionUpsertMetadataManager.shouldRevertMetadataOnInconsistency(_realtimeSegment); + } + protected long now() { return System.currentTimeMillis(); } From 83e4d27d91714230b367e4918a3ec5bd4a61c383 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Mon, 12 Oct 2026 14:05:59 -0700 Subject: [PATCH 05/13] Trim the holdsConsumerSemaphoreForMetadata() comment to what the code does not show --- .../realtime/RealtimeSegmentDataManager.java | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 48a198d6a306..6fe7a86e10f4 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1790,17 +1790,10 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) _realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata); } - /// Whether replacing this segment rewrites the key map its consuming rows wrote, so the next consuming segment - /// must not start until the replace is done. Such tables keep the consumer semaphore across both the build and the - /// download, which is what DISALLOW_ALWAYS does. - /// - /// Upsert: the replace reverts the metadata the consuming rows wrote, and the next segment would start against a - /// partially reverted primary key map. Dedup: a replica that consumed past the committed end offset downloads the - /// committed copy, and offloading the mutable segment removes the keys those extra rows added; a next consuming - /// segment running in between replays the rows, finds the keys still pointing at the mutable segment and drops - /// them as duplicates, so they land in neither segment. Dedup is not keyed on allowDedupConsumptionDuringCommit: - /// that one is deprecated and only read when no parallelSegmentConsumptionPolicy is set, so an explicit - /// ALLOW_ALWAYS would skip the hold it needs. + /// Whether replacing this segment rewrites the key map, which the next consuming segment must not run against. + /// These tables hold the semaphore across both the build and the download, the way DISALLOW_ALWAYS does. + /// Upsert reverts the keys the consuming rows wrote; dedup removes them when the mutable segment is offloaded, + /// and a replay in between is dropped as a duplicate against a key that is about to disappear. private boolean holdsConsumerSemaphoreForMetadata() { return revertsUpsertMetadata() || _partitionDedupMetadataManager != null; } From 6d027134a42e696b0c499c6736c8950dbae96a81 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Mon, 12 Oct 2026 14:12:36 -0700 Subject: [PATCH 06/13] Cut the semaphore hold comment to two lines --- .../data/manager/realtime/RealtimeSegmentDataManager.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 6fe7a86e10f4..906f9f814ead 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1790,10 +1790,8 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) _realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata); } - /// Whether replacing this segment rewrites the key map, which the next consuming segment must not run against. - /// These tables hold the semaphore across both the build and the download, the way DISALLOW_ALWAYS does. - /// Upsert reverts the keys the consuming rows wrote; dedup removes them when the mutable segment is offloaded, - /// and a replay in between is dropped as a duplicate against a key that is about to disappear. + /// The replace rewrites the key map, so hold the semaphore across the build and the download like + /// DISALLOW_ALWAYS: upsert reverts those keys, dedup removes them when the mutable segment is offloaded. private boolean holdsConsumerSemaphoreForMetadata() { return revertsUpsertMetadata() || _partitionDedupMetadataManager != null; } From 05587771257bd77fe800e44efd66f384acdcdd89 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Tue, 13 Oct 2026 21:24:24 -0700 Subject: [PATCH 07/13] Drop the pauseless exclusion from isTableTypeInconsistentDuringConsumption Excluding pauseless tables also took away their PROTECTED revert, so a reset of a pauseless consuming segment deleted the key instead of pointing it back at the committed segment, and the replay re-merged from nothing. It also changed what getParallelConsumptionPolicy() derives for a pauseless partial-upsert table with no explicit policy. The semaphore holds are unaffected. --- .../apache/pinot/segment/local/upsert/UpsertContext.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java index d1b653bb2269..625da707f943 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertContext.java @@ -26,7 +26,6 @@ import javax.annotation.Nullable; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.builder.ToStringBuilder; -import org.apache.pinot.common.utils.PauselessConsumptionUtils; import org.apache.pinot.segment.local.data.manager.TableDataManager; import org.apache.pinot.spi.config.table.HashFunction; import org.apache.pinot.spi.config.table.TableConfig; @@ -202,11 +201,8 @@ public File getTableIndexDir() { /// during segment replacement after force commit. This happens when: /// - Partial upsert is enabled (records need to be merged with previous values) /// - dropOutOfOrderRecord is enabled with NONE consistency mode (records may have been dropped) - /// Pauseless tables are excluded: they consume while the previous segment is still building, so the upsert - /// metadata cannot be kept reconcilable across replicas. public boolean isTableTypeInconsistentDuringConsumption() { - return (_dropOutOfOrderRecord || _outOfOrderRecordColumn != null || _partialUpsertHandlerSupplier != null) - && !PauselessConsumptionUtils.isPauselessEnabled(_tableConfig); + return _dropOutOfOrderRecord || _outOfOrderRecordColumn != null || _partialUpsertHandlerSupplier != null; } @Override From 1f820f11b86e212c1c85519b736b9a832df6f2f5 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Tue, 13 Oct 2026 21:38:06 -0700 Subject: [PATCH 08/13] Leave the build-time semaphore release alone Holding it across the build would make a pauseless table behave as DISALLOW_ALWAYS for every dedup table and, now that the pauseless exclusion is gone, for pauseless partial upsert in PROTECTED mode too. Keep the hold to the download, where the replace that rewrites the key map actually happens. --- .../data/manager/realtime/RealtimeSegmentDataManager.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 906f9f814ead..84945bb33148 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1207,9 +1207,7 @@ SegmentBuildDescriptor getSegmentBuildDescriptor() { @Nullable protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) throws SegmentBuildFailureException { - // Not released when the replace rewrites upsert or dedup metadata: a build that fails or mismatches the - // committed CRC falls back to a download, which cannot hold back a semaphore already given away. - if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild() && !holdsConsumerSemaphoreForMetadata()) { + if (_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()) { closeStreamConsumerAndReleaseSemaphore(); } // Do not allow building segment when table data manager is already shut down @@ -1790,8 +1788,8 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) _realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata); } - /// The replace rewrites the key map, so hold the semaphore across the build and the download like - /// DISALLOW_ALWAYS: upsert reverts those keys, dedup removes them when the mutable segment is offloaded. + /// 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. private boolean holdsConsumerSemaphoreForMetadata() { return revertsUpsertMetadata() || _partitionDedupMetadataManager != null; } From ba1dd6367c198c9a765e5002c4a5c60f02f095f7 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Wed, 14 Oct 2026 00:22:40 -0700 Subject: [PATCH 09/13] Hold the consumer semaphore through the build when the replace reverts upsert metadata The semaphore was released at the top of buildSegmentInternal() whenever the policy allowed consumption during build, regardless of what the replace that build leads to would do. For a table whose replace reverts upsert metadata that lets the next consuming segment start against the key map before the revert rewrites it -- the same hazard this PR already closes on the download side, and the one ALLOW_DURING_BUILD_ONLY reaches without any download at all. Guarded with revertsUpsertMetadata() rather than holdsConsumerSemaphoreForMetadata(): dedup 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 and it should not pay the serialization. This also makes the error branch below honest. It reads "the build holds the semaphore for these tables, so reaching this means it was released on another path", which until now was reachable through the ordinary path under ALLOW_ALWAYS and ALLOW_DURING_BUILD_ONLY. Co-Authored-By: Claude Opus 5 (1M context) --- .../data/manager/realtime/RealtimeSegmentDataManager.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 84945bb33148..b8fc458cab7e 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/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()) { closeStreamConsumerAndReleaseSemaphore(); } // Do not allow building segment when table data manager is already shut down From e98990d1de15f29c35fae45e7b2a29f05fcc14bd Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Thu, 15 Oct 2026 00:07:21 -0700 Subject: [PATCH 10/13] Report an overridden policy, and test the build-side guard Three things from review. The error log and UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD meter only fire when releasedDuringBuild is true, which is false by construction for ALLOW_DURING_DOWNLOAD_ONLY. Such a table was silently overridden with no signal. A second branch warns when the policy would have allowed parallel consumption and the semaphore is held anyway: nothing is at risk, so no meter, but the misconfiguration is now visible. The dedup rationale was wrong. Dedup re-points every key from the mutable segment to the new immutable one at the replace, not at offload, so it shares the build window rather than having none. It is still excluded, now as a cost decision rather than an absence of hazard. The guard had no test: FakeRealtimeSegmentDataManager replaces buildSegmentInternal wholesale, so an inlined condition was never reached. Extracted as shouldReleaseSemaphoreBeforeBuild(), routed the fake through it, and added the ALLOW_DURING_DOWNLOAD_ONLY case. The existing ALLOW_DURING_BUILD_ONLY assertion flips: the permit is now kept and nothing is flagged. Co-Authored-By: Claude Opus 5 (1M context) --- .../realtime/RealtimeSegmentDataManager.java | 22 +++++++++++----- .../RealtimeSegmentDataManagerTest.java | 26 ++++++++++++++++--- 2 files changed, 37 insertions(+), 11 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index b8fc458cab7e..6376ef026045 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1207,10 +1207,8 @@ SegmentBuildDescriptor getSegmentBuildDescriptor() { @Nullable protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) throws SegmentBuildFailureException { - // 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()) { + // Dedup shares this window but is left out: holding through every dedup build costs more than it buys. + if (shouldReleaseSemaphoreBeforeBuild()) { closeStreamConsumerAndReleaseSemaphore(); } // Do not allow building segment when table data manager is already shut down @@ -1778,12 +1776,17 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) 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. + // Already released elsewhere, so holding it now buys nothing. This is the case that can drop rows. _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); } + } else if (_parallelSegmentConsumptionPolicy.isAllowedDuringDownload() && holdsConsumerSemaphore) { + // Nothing at risk, but the table is misconfigured; the branch above cannot report it for this policy. + _segmentLogger.warn("Holding the consumer semaphore through this replace although the policy is {}: the replace " + + "rewrites upsert or dedup metadata. Set DISALLOW_ALWAYS to make this explicit.", + _parallelSegmentConsumptionPolicy); } if (allowedDuringDownload) { closeStreamConsumerAndReleaseSemaphore(); @@ -1791,8 +1794,13 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) _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. + /// Upsert reverts the keys at the replace; dedup re-points them to the new immutable segment. + /// Extracted to be testable: the test double replaces [#buildSegmentInternal] wholesale. + @VisibleForTesting + boolean shouldReleaseSemaphoreBeforeBuild() { + return _parallelSegmentConsumptionPolicy.isAllowedDuringBuild() && !revertsUpsertMetadata(); + } + private boolean holdsConsumerSemaphoreForMetadata() { return revertsUpsertMetadata() || _partitionDedupMetadataManager != null; } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java index 5ab454edf28c..64f1b96efb97 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java @@ -987,12 +987,29 @@ public void testDownloadAfterLocalBuildFlagsUpsertRevertWhenConsumingDuringBuild ParallelSegmentConsumptionPolicy.ALLOW_DURING_BUILD_ONLY, serverMetrics)) { Semaphore semaphore = _partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getSemaphore(); Assert.assertTrue(semaphore.tryAcquire()); + // The policy would release, but the replace reverts upsert metadata, so the guard keeps the permit. segmentDataManager._releaseSemaphoreOnBuild = true; + Assert.assertFalse(segmentDataManager.shouldReleaseSemaphoreBeforeBuild()); runGoOnlineForCrcGuard(segmentDataManager, crcMetadata(finalOffsetValue, 12345L), false, finalOffsetValue); Assert.assertTrue(segmentDataManager._downloadAndReplaceCalled); - Assert.assertEquals(semaphore.availablePermits(), 1); - verify(serverMetrics).addMeteredTableValue(anyString(), - eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD), eq(1L)); + Assert.assertEquals(semaphore.availablePermits(), 0); + verify(serverMetrics, never()).addMeteredTableValue(anyString(), + eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD), anyLong()); + } + + // ALLOW_DURING_DOWNLOAD_ONLY: releasedDuringBuild is never true here, so only the override path can report it. + _partitionGroupIdToConsumerCoordinatorMap.remove(PARTITION_GROUP_ID); + serverMetrics = spy(new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry())); + try (FakeRealtimeSegmentDataManager segmentDataManager = createFakeSegmentManagerForUpsertRevert( + ParallelSegmentConsumptionPolicy.ALLOW_DURING_DOWNLOAD_ONLY, serverMetrics)) { + Semaphore semaphore = _partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getSemaphore(); + Assert.assertTrue(semaphore.tryAcquire()); + segmentDataManager.getConsumerSemaphoreAcquired().set(true); + Assert.assertFalse(segmentDataManager.shouldReleaseSemaphoreBeforeBuild()); + segmentDataManager.downloadSegmentAndReplace(crcMetadata(finalOffsetValue, 12345L)); + Assert.assertEquals(semaphore.availablePermits(), 0); + verify(serverMetrics, never()).addMeteredTableValue(anyString(), + eq(ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD), anyLong()); } // ALLOW_DURING_BUILD_ONLY without a local build: the download still holds the semaphore, so nothing is flagged. @@ -1891,7 +1908,8 @@ protected boolean isLocalSegmentCrcMatchingZk(SegmentZKMetadata committedSegment protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) { terminateLoopIfNecessary(); _buildSegmentCalled = true; - if (_releaseSemaphoreOnBuild) { + // Through the real predicate, so the guard is exercised rather than bypassed. + if (_releaseSemaphoreOnBuild && shouldReleaseSemaphoreBeforeBuild()) { closeStreamConsumerAndReleaseSemaphore(); } if (_failSegmentBuild) { From fffbf63b5af40ae75663d61190ba79dd064a3c69 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Thu, 15 Oct 2026 00:19:19 -0700 Subject: [PATCH 11/13] Hoist the shared condition out of the two report branches Both tested holdsConsumerSemaphore. Nesting them under it says what they have in common -- the semaphore is held for metadata reasons -- and leaves each branch to state only how it got there. Co-Authored-By: Claude Opus 5 (1M context) --- .../realtime/RealtimeSegmentDataManager.java | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 6376ef026045..89548963424a 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1775,18 +1775,20 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) boolean holdsConsumerSemaphore = holdsConsumerSemaphoreForMetadata(); boolean allowedDuringDownload = _parallelSegmentConsumptionPolicy.isAllowedDuringDownload() && !holdsConsumerSemaphore; - if (releasedDuringBuild && holdsConsumerSemaphore) { - // Already released elsewhere, so holding it now buys nothing. This is the case that can drop rows. - _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 (holdsConsumerSemaphore) { + if (releasedDuringBuild) { + // Already released elsewhere, so holding it now buys nothing. This is the case that can drop rows. + _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); + } + } else if (_parallelSegmentConsumptionPolicy.isAllowedDuringDownload()) { + // Held against the policy. Nothing at risk, but the table asks for something it does not get. + _segmentLogger.warn("Holding the consumer semaphore through this replace although the policy is {}. Set " + + "DISALLOW_ALWAYS to make this explicit.", _parallelSegmentConsumptionPolicy); } - } else if (_parallelSegmentConsumptionPolicy.isAllowedDuringDownload() && holdsConsumerSemaphore) { - // Nothing at risk, but the table is misconfigured; the branch above cannot report it for this policy. - _segmentLogger.warn("Holding the consumer semaphore through this replace although the policy is {}: the replace " - + "rewrites upsert or dedup metadata. Set DISALLOW_ALWAYS to make this explicit.", - _parallelSegmentConsumptionPolicy); } if (allowedDuringDownload) { closeStreamConsumerAndReleaseSemaphore(); From bda8cd3783d8c5a2c1f208faefc87fd41a7f0711 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Thu, 15 Oct 2026 00:30:23 -0700 Subject: [PATCH 12/13] One branch and one message for both reports The if/else existed only to carry a second wording. One condition covers both: the policy would let the next consuming segment run while this replace rewrites the key map, whether it already did or the guard stopped it. The meter still fires only when the semaphore was released, since that is the case rows are at risk in. Co-Authored-By: Claude Opus 5 (1M context) --- .../realtime/RealtimeSegmentDataManager.java | 20 +++++++------------ 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 89548963424a..9e4aee305727 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1775,19 +1775,13 @@ protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) boolean holdsConsumerSemaphore = holdsConsumerSemaphoreForMetadata(); boolean allowedDuringDownload = _parallelSegmentConsumptionPolicy.isAllowedDuringDownload() && !holdsConsumerSemaphore; - if (holdsConsumerSemaphore) { - if (releasedDuringBuild) { - // Already released elsewhere, so holding it now buys nothing. This is the case that can drop rows. - _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); - } - } else if (_parallelSegmentConsumptionPolicy.isAllowedDuringDownload()) { - // Held against the policy. Nothing at risk, but the table asks for something it does not get. - _segmentLogger.warn("Holding the consumer semaphore through this replace although the policy is {}. Set " - + "DISALLOW_ALWAYS to make this explicit.", _parallelSegmentConsumptionPolicy); + if (holdsConsumerSemaphore && (releasedDuringBuild + || _parallelSegmentConsumptionPolicy.isAllowedDuringDownload())) { + _segmentLogger.warn("Policy {} would let the next consuming segment run while this replace rewrites upsert or " + + "dedup metadata. Use DISALLOW_ALWAYS", _parallelSegmentConsumptionPolicy); + // Only when the semaphore was already released: the rows are actually at risk then. + if (releasedDuringBuild && revertsUpsertMetadata()) { + _serverMetrics.addMeteredTableValue(_clientId, ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L); } } if (allowedDuringDownload) { From e75ada085c80bf5252f2884e8c91af10490cfd51 Mon Sep 17 00:00:00 2001 From: Chaitanya Deepthi Date: Wed, 21 Oct 2026 17:14:53 -0700 Subject: [PATCH 13/13] Hold the permit for non-pauseless dedup builds too, and simplify the download path A dedup table released the permit before its build, so a build that failed or mismatched the committed CRC fell back to a download with the permit already gone. Non-pauseless dedup now keeps it; pauseless dedup still releases, since holding through its build is what pauseless exists to avoid and it skips the CRC check. downloadSegmentAndReplace loses two locals: the release decision and the report are separate statements, and the report fires only when the permit is actually missing under a policy that releases it, so a successful hold no longer logs. --- .../realtime/RealtimeSegmentDataManager.java | 48 ++++++++++--------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 9e4aee305727..df626b8e6c0f 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -1207,7 +1207,6 @@ SegmentBuildDescriptor getSegmentBuildDescriptor() { @Nullable protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) throws SegmentBuildFailureException { - // Dedup shares this window but is left out: holding through every dedup build costs more than it buys. if (shouldReleaseSemaphoreBeforeBuild()) { closeStreamConsumerAndReleaseSemaphore(); } @@ -1766,37 +1765,42 @@ public void goOnlineFromConsuming(SegmentZKMetadata segmentZKMetadata) protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata) throws Exception { - // 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(); - // 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 (holdsConsumerSemaphore && (releasedDuringBuild - || _parallelSegmentConsumptionPolicy.isAllowedDuringDownload())) { - _segmentLogger.warn("Policy {} would let the next consuming segment run while this replace rewrites upsert or " - + "dedup metadata. Use DISALLOW_ALWAYS", _parallelSegmentConsumptionPolicy); - // Only when the semaphore was already released: the rows are actually at risk then. - if (releasedDuringBuild && revertsUpsertMetadata()) { + boolean holdsPermit = holdsConsumerSemaphoreForMetadata(); + if (_parallelSegmentConsumptionPolicy.isAllowedDuringDownload() && !holdsPermit) { + closeStreamConsumerAndReleaseSemaphore(); + } + // Under a policy that releases somewhere, the permit should still be ours, so a missing one means the next + // consuming segment is already running. DISALLOW_ALWAYS never releases, so a missing permit there just means this + // segment never consumed. + boolean policyReleasesSomewhere = _parallelSegmentConsumptionPolicy.isAllowedDuringBuild() + || _parallelSegmentConsumptionPolicy.isAllowedDuringDownload(); + if (holdsPermit && policyReleasesSomewhere && !_consumerSemaphoreAcquired.get()) { + _segmentLogger.error("Next consuming segment is running while this replace rewrites upsert or dedup metadata, " + + "so its rows can be dropped against keys this replace removes. Policy: {}. Use DISALLOW_ALWAYS", + _parallelSegmentConsumptionPolicy); + if (revertsUpsertMetadata()) { _serverMetrics.addMeteredTableValue(_clientId, ServerMeter.UPSERT_REVERT_WITH_CONSUMPTION_DURING_DOWNLOAD, 1L); } } - if (allowedDuringDownload) { - closeStreamConsumerAndReleaseSemaphore(); - } _realtimeTableDataManager.downloadAndReplaceConsumingSegment(segmentZKMetadata); } - /// Upsert reverts the keys at the replace; dedup re-points them to the new immutable segment. - /// Extracted to be testable: the test double replaces [#buildSegmentInternal] wholesale. + /// A build that fails or mismatches the committed CRC falls back to a download, and the permit cannot be taken back + /// then, so tables whose replace rewrites the key map keep it. Pauseless dedup is the exception: it is the one case + /// where holding costs what pauseless exists to provide, and it skips the CRC check, so only an outright build + /// failure leaves it exposed. + /// + /// Negation of [#holdsConsumerSemaphoreForMetadata()] plus that exception; keep the two in step. @VisibleForTesting boolean shouldReleaseSemaphoreBeforeBuild() { - return _parallelSegmentConsumptionPolicy.isAllowedDuringBuild() && !revertsUpsertMetadata(); + if (!_parallelSegmentConsumptionPolicy.isAllowedDuringBuild()) { + return false; + } + return !revertsUpsertMetadata() + && (_partitionDedupMetadataManager == null || PauselessConsumptionUtils.isPauselessEnabled(_tableConfig)); } + /// The replace re-points or reverts every key of the consuming segment, so the next one must not run against it. private boolean holdsConsumerSemaphoreForMetadata() { return revertsUpsertMetadata() || _partitionDedupMetadataManager != null; }