You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by an...@apache.org on 2025/11/16 16:45:29 UTC
(pinot) branch master updated: Fix Segment Partition Tracking for Multi-Topic Ingestion (#17217)
This is an automated email from the ASF dual-hosted git repository.
ankitsultana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 80062305e3a Fix Segment Partition Tracking for Multi-Topic Ingestion (#17217)
80062305e3a is described below
commit 80062305e3adc0f822123fcc2e52e680be1f336d
Author: tarun11Mavani <35...@users.noreply.github.com>
AuthorDate: Sun Nov 16 22:15:22 2025 +0530
Fix Segment Partition Tracking for Multi-Topic Ingestion (#17217)
---
.../helix/core/realtime/PinotLLCRealtimeSegmentManager.java | 10 +++++++---
1 file changed, 7 insertions(+), 3 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index a41eb37638b..ebbd8c9e5c2 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -1061,9 +1061,13 @@ public class PinotLLCRealtimeSegmentManager {
+ "Please update the table config accordingly.", numPartitionGroups,
columnPartitionConfig.getNumPartitions());
}
+ // For multi-stream tables, convert Pinot partition ID (which includes padding offset) to stream partition ID.
+ // This ensures the partition metadata stored in ZK matches what the broker's partition function computes
+ // during query pruning. For example, stream 1 partition 5 has Pinot partition ID 10005, but should store 5.
+ int streamPartitionId = IngestionConfigUtils.getStreamPartitionIdFromPinotPartitionId(partitionId);
ColumnPartitionMetadata columnPartitionMetadata =
new ColumnPartitionMetadata(columnPartitionConfig.getFunctionName(), numPartitionGroups,
- Collections.singleton(partitionId), columnPartitionConfig.getFunctionConfig());
+ Collections.singleton(streamPartitionId), columnPartitionConfig.getFunctionConfig());
return new SegmentPartitionMetadata(Collections.singletonMap(entry.getKey(), columnPartitionMetadata));
} else {
LOGGER.warn(
@@ -2402,9 +2406,9 @@ public class PinotLLCRealtimeSegmentManager {
Set<String> invalidSegments = partitionedByIsConsuming.get(false);
if (!invalidSegments.isEmpty()) {
LOGGER.warn("Cannot commit segments that are not in CONSUMING state. All consuming segments: {}, "
- + "provided segments to commit: {}. Ignoring all non-consuming segments, sampling 10: {}",
+ + "provided segments to commit: {}. Ignoring all non-consuming segments, sampling 10: {}",
allConsumingSegments,
- segmentsToCommitStr, invalidSegments.stream().limit(10).collect(Collectors.toSet()));
+ segmentsToCommitStr, invalidSegments.stream().limit(10).collect(Collectors.toSet()));
}
return validSegmentsToCommit;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org