Copilot commented on code in PR #18744:
URL: https://github.com/apache/pinot/pull/18744#discussion_r3405540615
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/routing/WorkerManager.java:
##########
@@ -904,6 +904,12 @@ private void
assignWorkersToPartitionedLeafFragment(DispatchablePlanMetadata met
// verifies that the partition table obtained from routing manager is
compatible with the hint options
checkPartitionInfoMap(partitionTableInfo, tableName, partitionKey,
partitionFunction, numWorkers);
+ // Attach unavailable segments (no available replica, excluded from the
partition info) to the metadata so that
+ // they are reported in the query response
Review Comment:
This comment implies all segments with no available replica are attached
here, but segments can be excluded from the partition info for other reasons
(e.g. pending segments that would reduce fully replicated servers). Consider
rewording to reflect that this is attaching the routing manager's explicit
"unavailable segments" list.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java:
##########
@@ -44,14 +44,21 @@
import org.slf4j.LoggerFactory;
-/**
- * The {@code PartitionDataManager} manages partitions of a table. It manages
- * 1. all the online segments associated with the partition and their
allocated servers
- * 2. all the replica of a specific segment.
- * It provides API to query
- * 1. For each partition ID, what are the servers that contains ALL segments
belong to this partition ID.
- * 2. For each server, what are all the partition IDs and list of segments
of those partition IDs on this server.
- */
+/// The `PartitionDataManager` manages partitions of a table. It manages
+/// 1. all the online segments associated with the partition and their
allocated servers
+/// 2. all the replica of a specific segment.
+/// It provides API to query
+/// 1. For each partition ID, what are the servers that contains ALL
segments belong to this partition ID.
+/// 2. For each server, what are all the partition IDs and list of segments
of those partition IDs on this server.
+///
+/// When computing the fully replicated servers for a partition, segments that
have not been fully replicated per
+/// their latest ideal assignment (newly pushed segments, new consuming
segments, or segments uploaded under a new
+/// name to replace other segments) are excluded from the partition info until
all their replicas become available,
+/// so that they don't reduce the fully replicated servers while loading. This
protection is bounded by the configured
+/// new segment expiration time since the ideal assignment of the segment last
changed; after that the segment is
+/// treated as a regular segment, so that a replica failing to load doesn't
exclude the segment from being served
+/// forever. Segments without any available replica are also excluded because
they cannot be served regardless of the
+/// server picked, and are reported via
[TablePartitionReplicatedServersInfo#getUnavailableSegments()].
Review Comment:
The class-level comment says segments with no available replica are reported
via `getUnavailableSegments()`, but the current logic only reports segments
once they are no longer treated as pending. The comment should be clarified so
it matches the actual behavior.
##########
pinot-core/src/main/java/org/apache/pinot/core/routing/TablePartitionReplicatedServersInfo.java:
##########
@@ -69,6 +71,12 @@ public List<String> getSegmentsWithInvalidPartition() {
return _segmentsWithInvalidPartition;
}
+ /// Returns the segments excluded from the partition info because they have
no available replica, thus cannot be
+ /// served regardless of the server picked.
Review Comment:
`getUnavailableSegments()` is documented as "segments excluded ... because
they have no available replica", which can be read as an exhaustive definition.
Consider clarifying that this is the set reported by the routing layer (and may
not include segments excluded for other reasons, such as
pending/not-yet-fully-replicated segments).
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java:
##########
@@ -44,14 +44,21 @@
import org.slf4j.LoggerFactory;
-/**
- * The {@code PartitionDataManager} manages partitions of a table. It manages
- * 1. all the online segments associated with the partition and their
allocated servers
- * 2. all the replica of a specific segment.
- * It provides API to query
- * 1. For each partition ID, what are the servers that contains ALL segments
belong to this partition ID.
- * 2. For each server, what are all the partition IDs and list of segments
of those partition IDs on this server.
- */
+/// The `PartitionDataManager` manages partitions of a table. It manages
Review Comment:
The class-level documentation refers to `PartitionDataManager`, but this
class is `SegmentPartitionMetadataManager`. This makes the Javadoc misleading
when browsing the code.
##########
pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java:
##########
@@ -285,6 +288,153 @@ public void
testPartitionMetadataManagerProcessingThroughSegmentChangesSinglePar
assertEquals(tablePartitionReplicatedServersInfo.getSegmentsWithInvalidPartition().get(0),
segmentInvalid);
}
+ @Test
+ public void testReplicationStateTracking() {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> externalViewAssignment =
externalView.getRecord().getMapFields();
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> idealStateAssignment =
idealState.getRecord().getMapFields();
+ Map<String, String> onlineInstanceStateMap = Map.of(SERVER_0, ONLINE,
SERVER_1, ONLINE);
+ 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);
+
+ // segmentA is fully replicated on both servers
+ String segmentA = "trackingSegmentA";
+ onlineSegments.add(segmentA);
+ idealStateAssignment.put(segmentA, onlineInstanceStateMap);
+ externalViewAssignment.put(segmentA, onlineInstanceStateMap);
+ setSegmentZKMetadata(segmentA, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ TablePartitionReplicatedServersInfo tablePartitionReplicatedServersInfo =
+ partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ TablePartitionReplicatedServersInfo.PartitionInfo[] partitionInfoMap =
+ tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
ImmutableSet.of(SERVER_0, SERVER_1));
+ assertEquals(partitionInfoMap[0]._segments,
Collections.singleton(segmentA));
+
+ // segmentB is old by creation time, but has never been fully replicated
per its ideal assignment (only ONLINE on
+ // SERVER_1 while assigned to both servers). It should be excluded from
the partition info instead of reducing the
+ // fully replicated servers, regardless of its creation time.
+ String segmentB = "trackingSegmentB";
+ onlineSegments.add(segmentB);
+ idealStateAssignment.put(segmentB, onlineInstanceStateMap);
+ externalViewAssignment.put(segmentB, Map.of(SERVER_1, ONLINE));
+ setSegmentZKMetadata(segmentB, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfoMap =
tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
ImmutableSet.of(SERVER_0, SERVER_1));
+ assertEquals(partitionInfoMap[0]._segments,
Collections.singleton(segmentA));
+
assertTrue(tablePartitionReplicatedServersInfo.getUnavailableSegments().isEmpty());
+
+ // Once segmentB becomes fully replicated, it should be included as a
regular segment
+ externalViewAssignment.put(segmentB, onlineInstanceStateMap);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfoMap =
tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
ImmutableSet.of(SERVER_0, SERVER_1));
+ assertEqualsNoOrder(partitionInfoMap[0]._segments.toArray(), new
String[]{segmentA, segmentB});
+
+ // Losing a replica after having been fully replicated should reduce the
fully replicated servers, so that queries
+ // are routed to the remaining replica
+ externalViewAssignment.put(segmentB, Map.of(SERVER_0, ONLINE, SERVER_1,
ERROR));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfoMap =
tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
Collections.singleton(SERVER_0));
+ assertEqualsNoOrder(partitionInfoMap[0]._segments.toArray(), new
String[]{segmentA, segmentB});
+
+ // Losing all replicas should exclude the segment from the partition info
and report it as unavailable, instead of
+ // clearing the fully replicated servers and failing all the queries on
the partition
+ externalViewAssignment.put(segmentB, Map.of(SERVER_0, ERROR, SERVER_1,
ERROR));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfoMap =
tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
ImmutableSet.of(SERVER_0, SERVER_1));
+ assertEquals(partitionInfoMap[0]._segments,
Collections.singleton(segmentA));
+ assertEquals(tablePartitionReplicatedServersInfo.getUnavailableSegments(),
+ Collections.singletonList(segmentB));
+
+ // Recovering all replicas should include the segment back as a regular
segment
+ externalViewAssignment.put(segmentB, onlineInstanceStateMap);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfoMap =
tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
ImmutableSet.of(SERVER_0, SERVER_1));
+ assertEqualsNoOrder(partitionInfoMap[0]._segments.toArray(), new
String[]{segmentA, segmentB});
+
assertTrue(tablePartitionReplicatedServersInfo.getUnavailableSegments().isEmpty());
+
+ // Changing the ideal assignment (e.g. rebalance) should reset the
replication state. While moving towards the new
+ // assignment, the segment should not reduce the fully replicated servers,
but should still be included when its
+ // online servers cover them.
+ idealStateAssignment.put(segmentB, Map.of(SERVER_0, ONLINE, SERVER_2,
ONLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfoMap =
tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
ImmutableSet.of(SERVER_0, SERVER_1));
+ assertEqualsNoOrder(partitionInfoMap[0]._segments.toArray(), new
String[]{segmentA, segmentB});
+
+ // Once the segment converges to the new assignment, it should reduce the
fully replicated servers as a regular
+ // segment
+ externalViewAssignment.put(segmentB, Map.of(SERVER_0, ONLINE, SERVER_2,
ONLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfoMap =
tablePartitionReplicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
Collections.singleton(SERVER_0));
+ assertEqualsNoOrder(partitionInfoMap[0]._segments.toArray(), new
String[]{segmentA, segmentB});
+ }
+
+ @Test
+ public void testPendingSegmentExpiration()
+ throws InterruptedException {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> externalViewAssignment =
externalView.getRecord().getMapFields();
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> idealStateAssignment =
idealState.getRecord().getMapFields();
+ Map<String, String> onlineInstanceStateMap = Map.of(SERVER_0, ONLINE,
SERVER_1, ONLINE);
+ Set<String> onlineSegments = new HashSet<>();
+
+ // Use 0 expiration time so that the pending state expires right away
+ SegmentPartitionMetadataManager partitionMetadataManager =
+ new SegmentPartitionMetadataManager(OFFLINE_TABLE_NAME,
PARTITION_COLUMN, PARTITION_COLUMN_FUNC, NUM_PARTITIONS,
+ 0L);
+ SegmentZkMetadataFetcher segmentZkMetadataFetcher =
+ new SegmentZkMetadataFetcher(OFFLINE_TABLE_NAME, _propertyStore);
+ segmentZkMetadataFetcher.register(partitionMetadataManager);
+ segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
+
+ // segmentX is fully replicated on both servers; segmentY is assigned to
both servers but only ONLINE on SERVER_1
+ String segmentX = "expirationSegmentX";
+ onlineSegments.add(segmentX);
+ idealStateAssignment.put(segmentX, onlineInstanceStateMap);
+ externalViewAssignment.put(segmentX, onlineInstanceStateMap);
+ setSegmentZKMetadata(segmentX, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ String segmentY = "expirationSegmentY";
+ onlineSegments.add(segmentY);
+ idealStateAssignment.put(segmentY, onlineInstanceStateMap);
+ externalViewAssignment.put(segmentY, Map.of(SERVER_1, ONLINE));
+ setSegmentZKMetadata(segmentY, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+
+ // After the pending state expires, segmentY should be treated as a
regular segment and reduce the fully
+ // replicated servers, so that its data is still served from the available
replica
+ Thread.sleep(10);
Review Comment:
Using `Thread.sleep(10)` to advance the clock can be flaky on slow/loaded CI
machines and makes the test timing-dependent. A deterministic wait for the
system clock to tick avoids spurious failures while still validating the
expiration behavior.
--
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]
