This is an automated email from the ASF dual-hosted git repository.
raghavyadav01 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 2dc0baa9a8b Defer a new segment only while its replicas are still
coming up (#19901)
2dc0baa9a8b is described below
commit 2dc0baa9a8b220716c8fab8a02cc9395e92beb6c
Author: RAGHVENDRA KUMAR YADAV <[email protected]>
AuthorDate: Mon Oct 19 14:30:06 2026 -0700
Defer a new segment only while its replicas are still coming up (#19901)
* Defer a new segment only while its replicas are still coming up
The broker's partition info leaves a newly created segment out of its
partition
until the segment is online on every server that already holds the
partition,
so that a segment whose replicas are still loading does not shrink the
partition's fully replicated servers. The same test also excludes a segment
that
is already online on every server it is assigned to, just on different
servers
than the rest of its partition. That segment never covers them, so it stays
out
for the whole new-segment window, and a partition-hinted multi-stage query
silently skips its rows: two partitioned tables without partition-aware
placement returned 23 of 60 rows for a colocated join, with no error. Once
the
segment stops being new the same query fails with a clear error instead.
Track the replica count the ideal state assigns, and defer a new segment
only
while fewer replicas are online. A fully online segment is counted like any
other, so a partition that no server holds whole is reported at once.
* Judge a new segment's readiness by its assigned servers, not their count
The ideal state and external view are updated independently, so during a
reassignment the external view can still list a server the ideal state has
dropped. Comparing replica counts then admits a segment while an assigned
replica
is still down. Track the assigned server list and require every assigned
server
to be online. Also rename the change flag, which now covers assignment
changes,
and note why an empty assignment keeps deferring.
---
.../SegmentPartitionMetadataManager.java | 65 +++++++++++++++----
.../SegmentPartitionMetadataManagerTest.java | 75 ++++++++++++++++++++++
2 files changed, 128 insertions(+), 12 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
index cf41e32ee1d..d968e59cbff 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
@@ -112,7 +112,8 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
String segment = onlineSegments.get(i);
ZNRecord znRecord = znRecords.get(i);
SegmentInfo segmentInfo = new SegmentInfo(getPartitionId(segment,
znRecord), getCreationTimeMs(znRecord),
- getOnlineServers(externalView.getStateMap(segment),
onlineServersCache));
+ getOnlineServers(externalView.getStateMap(segment),
onlineServersCache),
+ getOnlineServers(idealState.getInstanceStateMap(segment),
onlineServersCache));
_segmentInfoMap.put(segment, segmentInfo);
}
computeAllTablePartitionInfo();
@@ -148,7 +149,8 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
return SegmentUtils.getSegmentCreationTimeMs(new
SegmentZKMetadata(znRecord));
}
- /// Returns the immutable list of servers hosting the segment in ONLINE or
CONSUMING state.
+ /// Returns the immutable list of servers in ONLINE or CONSUMING state in
`instanceStateMap`: the servers hosting the
+ /// segment for an external view map, the servers it is assigned to for an
ideal state map.
///
/// Equal lists are shared through `onlineServersCache`, so that the many
segments hosted on the same servers (e.g.
/// with replica-group assignment) keep a single list instead of one per
segment.
@@ -173,7 +175,9 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
return instanceState.equals(SegmentStateModel.ONLINE) ||
instanceState.equals(SegmentStateModel.CONSUMING);
}
- /// Returns whether `onlineServers` holds exactly the servers in ONLINE or
CONSUMING state in `instanceStateMap`.
+
+ /// Returns whether `onlineServers` holds exactly the servers in ONLINE or
CONSUMING state in `instanceStateMap` (of
+ /// either the external view or the ideal state).
/// Does not allocate, so that the common case of an unchanged segment costs
no garbage.
private static boolean hasSameOnlineServers(List<String> onlineServers,
@Nullable Map<String, String> instanceStateMap) {
@@ -200,7 +204,9 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
// and their partition ids, TablePartitionReplicatedServersInfo also
depends on the online servers of each segment
// and on the clock (see _nextNewSegmentExpirationTimeMs).
boolean partitionsChanged = false;
- boolean onlineServersChanged = false;
+ // Set when the online (external view) or assigned (ideal state) servers
of a segment change: either can decide
+ // whether a new segment is still waiting for replicas.
+ boolean serversChanged = false;
Map<List<String>, List<String>> onlineServersCache = new HashMap<>();
// Update segment partition id for the pulled segments
// NOTE: Segments without ZK metadata are pulled again on every change, so
compare with the existing info.
@@ -213,7 +219,8 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
SegmentInfo segmentInfo = _segmentInfoMap.get(segment);
if (segmentInfo == null) {
_segmentInfoMap.put(segment, new SegmentInfo(partitionId,
creationTimeMs,
- getOnlineServers(externalView.getStateMap(segment),
onlineServersCache)));
+ getOnlineServers(externalView.getStateMap(segment),
onlineServersCache),
+ getOnlineServers(idealState.getInstanceStateMap(segment),
onlineServersCache)));
partitionsChanged = true;
} else if (segmentInfo._partitionId != partitionId ||
segmentInfo._creationTimeMs != creationTimeMs) {
segmentInfo._partitionId = partitionId;
@@ -230,12 +237,20 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
LOGGER.error("Failed to find segment info for segment: {} in table: {}
while handling assignment change",
segment, _tableNameWithType);
segmentInfo = new SegmentInfo(INVALID_PARTITION_ID,
INVALID_CREATION_TIME_MS,
- getOnlineServers(instanceStateMap, onlineServersCache));
+ getOnlineServers(instanceStateMap, onlineServersCache),
+ getOnlineServers(idealState.getInstanceStateMap(segment),
onlineServersCache));
_segmentInfoMap.put(segment, segmentInfo);
partitionsChanged = true;
- } else if (!hasSameOnlineServers(segmentInfo._onlineServers,
instanceStateMap)) {
- segmentInfo._onlineServers = getOnlineServers(instanceStateMap,
onlineServersCache);
- onlineServersChanged = true;
+ } else {
+ if (!hasSameOnlineServers(segmentInfo._onlineServers,
instanceStateMap)) {
+ segmentInfo._onlineServers = getOnlineServers(instanceStateMap,
onlineServersCache);
+ serversChanged = true;
+ }
+ Map<String, String> idealInstanceStateMap =
idealState.getInstanceStateMap(segment);
+ if (!hasSameOnlineServers(segmentInfo._assignedServers,
idealInstanceStateMap)) {
+ segmentInfo._assignedServers =
getOnlineServers(idealInstanceStateMap, onlineServersCache);
+ serversChanged = true;
+ }
}
}
if (_segmentInfoMap.keySet().retainAll(onlineSegments)) {
@@ -243,7 +258,7 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
}
if (partitionsChanged) {
computeAllTablePartitionInfo();
- } else if (onlineServersChanged || _clock.millis() >
_nextNewSegmentExpirationTimeMs) {
+ } else if (serversChanged || _clock.millis() >
_nextNewSegmentExpirationTimeMs) {
computeTablePartitionReplicatedServersInfo();
}
}
@@ -257,7 +272,7 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
// NOTE: This should not happen, but we still handle it gracefully by
adding an invalid SegmentInfo
LOGGER.error("Failed to find segment info for segment: {} in table: {}
while handling segment refresh", segment,
_tableNameWithType);
- segmentInfo = new SegmentInfo(partitionId, pushTimeMs, List.of());
+ segmentInfo = new SegmentInfo(partitionId, pushTimeMs, List.of(),
List.of());
_segmentInfoMap.put(segment, segmentInfo);
} else {
segmentInfo._partitionId = partitionId;
@@ -383,6 +398,17 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
//noinspection SlowListContainsAll
if
(onlineServers.containsAll(partitionInfo._fullyReplicatedServers)) {
partitionInfo._segments.add(segment);
+ } else if (segmentInfo.hasAllAssignedReplicasOnline()) {
+ // Every replica the ideal state assigns is already online, so the
segment is not waiting for replicas: it
+ // is placed on other servers than the rest of its partition.
Waiting would never change that, and
+ // excluding it would make any consumer of this info (e.g. a
partition-hinted multi-stage leaf) silently
+ // skip its rows until it stops being new. Count it like an old
segment instead, so the partition reports
+ // the servers that really hold all of it -- possibly none.
+ if
(partitionInfo._fullyReplicatedServers.retainAll(onlineServers)) {
+ segmentsReducingFullyReplicatedServers.add(
+ Triple.of(segment, partitionId,
partitionInfo._fullyReplicatedServers.size()));
+ }
+ partitionInfo._segments.add(segment);
} else {
excludedNewSegments.add(segment);
excludedNewSegmentPartitions.add(partitionId);
@@ -458,11 +484,26 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
int _partitionId;
long _creationTimeMs;
List<String> _onlineServers;
+ /// Servers the ideal state assigns the segment to in ONLINE or CONSUMING
state; empty when the ideal state does
+ /// not list it, or lists it only in other states (e.g. OFFLINE while it
is being dropped).
+ List<String> _assignedServers;
- SegmentInfo(int partitionId, long creationTimeMs, List<String>
onlineServers) {
+ SegmentInfo(int partitionId, long creationTimeMs, List<String>
onlineServers, List<String> assignedServers) {
_partitionId = partitionId;
_creationTimeMs = creationTimeMs;
_onlineServers = onlineServers;
+ _assignedServers = assignedServers;
+ }
+
+ /// Whether every server the ideal state assigns the segment to has it
online, i.e. it is not still coming up.
+ ///
+ /// Compares the server sets rather than their sizes: the ideal state and
external view are updated independently,
+ /// so during a reassignment the external view can still list a server the
ideal state has dropped, matching the
+ /// assigned count while an assigned replica is still down. An empty
assignment returns false, which keeps
+ /// deferring a segment whose readiness cannot be judged -- the behaviour
before assignments were tracked.
+ boolean hasAllAssignedReplicasOnline() {
+ //noinspection SlowListContainsAll
+ return !_assignedServers.isEmpty() &&
_onlineServers.containsAll(_assignedServers);
}
}
}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
index 5c6d6af0cbb..8be8c804aeb 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
@@ -295,6 +295,81 @@ public class SegmentPartitionMetadataManagerTest extends
ControllerTest {
assertEquals(tablePartitionReplicatedServersInfo.getSegmentsWithInvalidPartition().get(0),
segmentInvalid);
}
+ /// A new segment is deferred only while its replicas are still coming up.
One that is already online on every server
+ /// the ideal state assigns, but on other servers than the rest of its
partition, will never cover them: deferring it
+ /// made a partition-hinted query silently skip its rows until it stopped
being new. It must instead count like an
+ /// old segment, so the partition reports that no server holds all of it.
+ @Test
+ public void testFullyOnlineNewSegmentOnOtherServersIsNotDeferred() {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> idealAssignment =
idealState.getRecord().getMapFields();
+ Set<String> onlineSegments = new HashSet<>();
+
+ SegmentPartitionMetadataManager partitionMetadataManager =
+ new SegmentPartitionMetadataManager(OFFLINE_TABLE_NAME,
PARTITION_COLUMN, PARTITION_COLUMN_FUNC, NUM_PARTITIONS,
+ TimeUnit.MINUTES.toMillis(5));
+ SegmentZkMetadataFetcher segmentZkMetadataFetcher =
+ new SegmentZkMetadataFetcher(OFFLINE_TABLE_NAME, _propertyStore);
+ segmentZkMetadataFetcher.register(partitionMetadataManager);
+ segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
+
+ // An old segment of partition 0 on server0.
+ String oldSegment = "oldSegment";
+ onlineSegments.add(oldSegment);
+ idealAssignment.put(oldSegment, Map.of(SERVER_0, ONLINE));
+ segmentAssignment.put(oldSegment, Map.of(SERVER_0, ONLINE));
+ setSegmentZKMetadata(oldSegment, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+
+ // A new segment of partition 0 still coming up: assigned to server0 and
server1, online on server1 only.
+ String comingUp = "comingUpSegment";
+ onlineSegments.add(comingUp);
+ idealAssignment.put(comingUp, Map.of(SERVER_0, ONLINE, SERVER_1, ONLINE));
+ segmentAssignment.put(comingUp, Map.of(SERVER_1, ONLINE));
+ setSegmentZKMetadata(comingUp, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
System.currentTimeMillis());
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+
+ TablePartitionReplicatedServersInfo.PartitionInfo[] partitionInfoMap =
+
partitionMetadataManager.getTablePartitionReplicatedServersInfo().getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._segments, List.of(oldSegment), "a
segment still coming up stays deferred");
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
Set.of(SERVER_0));
+
+ // A new segment of partition 0 fully online, but only on server1 -- the
default spread of a table without
+ // partition-aware placement.
+ String placedElsewhere = "placedElsewhereSegment";
+ onlineSegments.add(placedElsewhere);
+ idealAssignment.put(placedElsewhere, Map.of(SERVER_1, ONLINE));
+ segmentAssignment.put(placedElsewhere, Map.of(SERVER_1, ONLINE));
+ setSegmentZKMetadata(placedElsewhere, PARTITION_COLUMN_FUNC,
NUM_PARTITIONS, 0, System.currentTimeMillis());
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+
+ partitionInfoMap =
partitionMetadataManager.getTablePartitionReplicatedServersInfo().getPartitionInfoMap();
+ assertEqualsNoOrder(partitionInfoMap[0]._segments.toArray(), new
String[]{oldSegment, placedElsewhere});
+ assertTrue(partitionInfoMap[0]._fullyReplicatedServers.isEmpty(),
+ "no server holds every segment of the partition, and the info must say
so rather than hide a segment");
+
+ // A new segment of partition 1 mid-reassignment: the ideal state moved it
to server0 and server1, while the
+ // external view still lists the server it is leaving next to server1. Two
online servers match the two assigned,
+ // yet assigned server0 is still down, so the segment is still coming up
and must stay deferred.
+ String oldSegment1 = "oldSegment1";
+ onlineSegments.add(oldSegment1);
+ idealAssignment.put(oldSegment1, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ segmentAssignment.put(oldSegment1, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ setSegmentZKMetadata(oldSegment1, PARTITION_COLUMN_FUNC, NUM_PARTITIONS,
1, 0L);
+ String midReassignment = "midReassignmentSegment";
+ onlineSegments.add(midReassignment);
+ idealAssignment.put(midReassignment, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ segmentAssignment.put(midReassignment, Map.of(SERVER_1, ONLINE,
"leavingServer", ONLINE));
+ setSegmentZKMetadata(midReassignment, PARTITION_COLUMN_FUNC,
NUM_PARTITIONS, 1, System.currentTimeMillis());
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+
+ partitionInfoMap =
partitionMetadataManager.getTablePartitionReplicatedServersInfo().getPartitionInfoMap();
+ assertEquals(partitionInfoMap[1]._segments, List.of(oldSegment1),
+ "an assigned replica is still down, so the segment is not ready even
though the replica counts match");
+ assertEquals(partitionInfoMap[1]._fullyReplicatedServers, Set.of(SERVER_0,
SERVER_1));
+ }
+
/// A partition whose only segments are new ones without all replicas
available holds data that no single server can
/// serve as a whole, so it must be told apart from a genuinely empty
partition (see
///
[TablePartitionReplicatedServersInfo#getPartitionsWithOnlyDeferredSegments()]).
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
