You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ra...@apache.org on 2026/10/21 01:58:09 UTC
(pinot) branch master updated: Add onlyIfNeeded option in segment reload (#19861)
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 b1d6f042e3f Add onlyIfNeeded option in segment reload (#19861)
b1d6f042e3f is described below
commit b1d6f042e3fb6f3cf614e9742b618b135cf222dd
Author: Jhow <44...@users.noreply.github.com>
AuthorDate: Wed Oct 21 09:58:03 2026 +0800
Add onlyIfNeeded option in segment reload (#19861)
* add onlyIfNeeded flag to segment reload
* fix duplicate count in skipped and success count
* reuse isReloadNeeded in needReloadSegments
* inline the single segment reload check
* nit: logs
* docs
---
.../common/messages/SegmentReloadMessage.java | 22 +++
.../common/messages/SegmentReloadMessageTest.java | 13 ++
.../api/resources/PinotTableReloadResource.java | 11 +-
.../helix/core/PinotHelixResourceManager.java | 42 ++++-
.../services/PinotTableReloadService.java | 24 +--
.../services/PinotTableReloadServiceTest.java | 94 +++++++---
.../core/data/manager/BaseTableDataManager.java | 169 +++++++++++++++---
.../core/data/manager/InstanceDataManager.java | 29 ++-
.../data/manager/BaseTableDataManagerTest.java | 198 +++++++++++++++++++++
.../local/data/manager/TableDataManager.java | 32 +++-
.../pinot/segment/local/utils/ReloadJobStatus.java | 20 +++
.../local/utils/ServerReloadJobStatusCache.java | 12 ++
.../utils/ServerReloadJobStatusCacheTest.java | 20 +++
.../api/resources/ControllerJobStatusResource.java | 19 +-
.../starter/helix/HelixInstanceDataManager.java | 14 +-
.../helix/SegmentMessageHandlerFactory.java | 10 +-
.../resources/ControllerJobStatusResourceTest.java | 97 ++++++++++
17 files changed, 734 insertions(+), 92 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/messages/SegmentReloadMessage.java b/pinot-common/src/main/java/org/apache/pinot/common/messages/SegmentReloadMessage.java
index 76b7548b5cc..67f10d66018 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/messages/SegmentReloadMessage.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/messages/SegmentReloadMessage.java
@@ -33,6 +33,7 @@ public class SegmentReloadMessage extends Message {
public static final String RELOAD_SEGMENT_MSG_SUB_TYPE = "RELOAD_SEGMENT";
private static final String FORCE_DOWNLOAD_KEY = "forceDownload";
+ private static final String ONLY_IF_NEEDED_KEY = "onlyIfNeeded";
private static final String SEGMENT_NAMES = "segmentNames";
private static final String RELOAD_JOB_ID_KEY = "reloadJobId";
@@ -61,6 +62,20 @@ public class SegmentReloadMessage extends Message {
/// @param reloadJobId reload job id to associate with the reload operation
public SegmentReloadMessage(String tableNameWithType, @Nullable List<String> segmentNames, boolean forceDownload,
@Nullable String reloadJobId) {
+ this(tableNameWithType, segmentNames, forceDownload, false, reloadJobId);
+ }
+
+ /// This msg asks server to reload a list of specified segments with an explicit reload job id.
+ ///
+ /// @param tableNameWithType the table where the segments are from.
+ /// @param segmentNames a list of specified segments to reload, or null for all segments.
+ /// @param forceDownload whether to download segments from deep store when reloading.
+ /// @param onlyIfNeeded whether to reload only the segments whose loaded copy is out of date (CRC change, tier
+ /// change or preprocessing needed). Ignored when `forceDownload` is true. Servers that do
+ /// not recognize this field reload every targeted segment.
+ /// @param reloadJobId reload job id to associate with the reload operation
+ public SegmentReloadMessage(String tableNameWithType, @Nullable List<String> segmentNames, boolean forceDownload,
+ boolean onlyIfNeeded, @Nullable String reloadJobId) {
super(MessageType.USER_DEFINE_MSG, UUID.randomUUID().toString());
setResourceName(tableNameWithType);
setMsgSubType(RELOAD_SEGMENT_MSG_SUB_TYPE);
@@ -76,6 +91,9 @@ public class SegmentReloadMessage extends Message {
}
znRecord.setBooleanField(FORCE_DOWNLOAD_KEY, forceDownload);
+ if (onlyIfNeeded) {
+ znRecord.setBooleanField(ONLY_IF_NEEDED_KEY, true);
+ }
if (CollectionUtils.isNotEmpty(segmentNames)) {
// TODO: use the new List field and deprecate the partition name in next release.
setPartitionName(segmentNames.get(0));
@@ -94,6 +112,10 @@ public class SegmentReloadMessage extends Message {
return getRecord().getBooleanField(FORCE_DOWNLOAD_KEY, false);
}
+ public boolean shouldReloadOnlyIfNeeded() {
+ return getRecord().getBooleanField(ONLY_IF_NEEDED_KEY, false);
+ }
+
@Nullable
public List<String> getSegmentList() {
return getRecord().getListField(SEGMENT_NAMES);
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/messages/SegmentReloadMessageTest.java b/pinot-common/src/test/java/org/apache/pinot/common/messages/SegmentReloadMessageTest.java
index 6f8ac6ca723..998d1ffd614 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/messages/SegmentReloadMessageTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/messages/SegmentReloadMessageTest.java
@@ -34,9 +34,22 @@ public class SegmentReloadMessageTest {
SegmentReloadMessage message = new SegmentReloadMessage("testTable_OFFLINE", false);
assertFalse(message.shouldForceDownload());
+ assertFalse(message.shouldReloadOnlyIfNeeded());
assertEquals(message.getReloadJobId(), message.getMsgId());
}
+ @Test
+ public void testOnlyIfNeededSurvivesRoundTrip() {
+ SegmentReloadMessage message =
+ new SegmentReloadMessage("testTable_OFFLINE", List.of("seg_1"), false, true, "job-123");
+ SegmentReloadMessage received = new SegmentReloadMessage(new Message(message.getRecord()));
+
+ assertTrue(received.shouldReloadOnlyIfNeeded());
+ assertFalse(received.shouldForceDownload());
+ assertEquals(received.getSegmentList(), List.of("seg_1"));
+ assertEquals(received.getReloadJobId(), "job-123");
+ }
+
@Test
public void testReloadJobIdOverrideAndSegments() {
List<String> segments = List.of("seg_1", "seg_2");
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableReloadResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableReloadResource.java
index d38d9542b3a..b8b40bc83ad 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableReloadResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableReloadResource.java
@@ -77,6 +77,9 @@ import static org.apache.pinot.spi.utils.CommonConstants.SWAGGER_AUTHORIZATION_K
@Path("/")
public class PinotTableReloadResource {
private static final Logger LOG = LoggerFactory.getLogger(PinotTableReloadResource.class);
+ private static final String ONLY_IF_NEEDED_DESCRIPTION = "Reload only the segments whose loaded copy is out of "
+ + "date (CRC changed, tier changed, or index/schema preprocessing needed); up-to-date segments are skipped and "
+ + "counted as successful. Ignored when forceDownload is true";
private final PinotTableReloadService _service;
private final PinotTableReloadStatusReporter _statusReporter;
@@ -106,9 +109,11 @@ public class PinotTableReloadResource {
@PathParam("segmentName") @Encoded String segmentName,
@ApiParam(value = "Force server to re-download segment from deep store", defaultValue = "false")
@QueryParam("forceDownload") @DefaultValue("false") boolean forceDownload,
+ @ApiParam(value = ONLY_IF_NEEDED_DESCRIPTION, defaultValue = "false")
+ @QueryParam("onlyIfNeeded") @DefaultValue("false") boolean onlyIfNeeded,
@ApiParam(value = "Target specific server instance") @QueryParam("targetInstance") @Nullable
String targetInstance, @Context HttpHeaders headers) {
- return _service.reloadSegment(tableName, segmentName, forceDownload, targetInstance, headers);
+ return _service.reloadSegment(tableName, segmentName, forceDownload, onlyIfNeeded, targetInstance, headers);
}
@POST
@@ -133,6 +138,8 @@ public class PinotTableReloadResource {
String tableTypeStr,
@ApiParam(value = "Force server to re-download segments from deep store", defaultValue = "false")
@QueryParam("forceDownload") @DefaultValue("false") boolean forceDownload,
+ @ApiParam(value = ONLY_IF_NEEDED_DESCRIPTION, defaultValue = "false")
+ @QueryParam("onlyIfNeeded") @DefaultValue("false") boolean onlyIfNeeded,
@ApiParam(value = "Start timestamp (inclusive) in milliseconds. Omit for an unbounded lower bound.")
@QueryParam("startTimestamp") String startTimestampStr,
@ApiParam(value = "End timestamp (exclusive) in milliseconds. Omit for an unbounded upper bound.")
@@ -144,7 +151,7 @@ public class PinotTableReloadResource {
@ApiParam(value = "JSON map of instance to segment lists (overrides targetInstance)")
@QueryParam("instanceToSegmentsMap") @Nullable String instanceToSegmentsMapInJson, @Context HttpHeaders headers)
throws IOException {
- return _service.reloadAllSegments(tableName, tableTypeStr, forceDownload, targetInstance,
+ return _service.reloadAllSegments(tableName, tableTypeStr, forceDownload, onlyIfNeeded, targetInstance,
instanceToSegmentsMapInJson, startTimestampStr, endTimestampStr, excludeOverlapping, headers);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index cbe7da8b0b3..2d536a21b42 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -3197,15 +3197,22 @@ public class PinotHelixResourceManager {
public Map<String, Integer> reloadSegments(String tableNameWithType, boolean forceDownload,
Map<String, List<String>> instanceToSegmentsMap, String reloadJobId) {
- LOGGER.info("Sending reload messages for table: {} with forceDownload: {}, and instanceToSegmentsMap: {}",
- tableNameWithType, forceDownload, instanceToSegmentsMap);
+ return reloadSegments(tableNameWithType, forceDownload, false, instanceToSegmentsMap, reloadJobId);
+ }
+
+ /// Sends reload messages for the given segments to their target instances. When `onlyIfNeeded` is true, servers
+ /// reload only the segments whose loaded copy is out of date and skip the rest.
+ public Map<String, Integer> reloadSegments(String tableNameWithType, boolean forceDownload, boolean onlyIfNeeded,
+ Map<String, List<String>> instanceToSegmentsMap, String reloadJobId) {
+ LOGGER.info("Sending reload messages for table: {} with forceDownload: {}, onlyIfNeeded: {}, and "
+ + "instanceToSegmentsMap: {}", tableNameWithType, forceDownload, onlyIfNeeded, instanceToSegmentsMap);
ClusterMessagingService messagingService = _helixZkManager.getMessagingService();
Map<String, Integer> instanceMsgInfoMap = new HashMap<>();
for (Map.Entry<String, List<String>> entry : instanceToSegmentsMap.entrySet()) {
String targetInstance = entry.getKey();
SegmentReloadMessage message =
- new SegmentReloadMessage(tableNameWithType, entry.getValue(), forceDownload, reloadJobId);
+ new SegmentReloadMessage(tableNameWithType, entry.getValue(), forceDownload, onlyIfNeeded, reloadJobId);
int numMessagesSent =
MessagingServiceUtils.send(messagingService, message, tableNameWithType, null, targetInstance);
if (numMessagesSent > 0) {
@@ -3221,11 +3228,19 @@ public class PinotHelixResourceManager {
public Pair<Integer, String> reloadAllSegments(String tableNameWithType, boolean forceDownload,
@Nullable String targetInstance) {
- LOGGER.info("Sending reload message for table: {} with forceDownload: {}, and target: {}", tableNameWithType,
- forceDownload, targetInstance == null ? "every instance" : targetInstance);
+ return reloadAllSegments(tableNameWithType, forceDownload, false, targetInstance);
+ }
+
+ /// Sends a reload message for all segments of the table. When `onlyIfNeeded` is true, servers reload only the
+ /// segments whose loaded copy is out of date and skip the rest.
+ public Pair<Integer, String> reloadAllSegments(String tableNameWithType, boolean forceDownload,
+ boolean onlyIfNeeded, @Nullable String targetInstance) {
+ LOGGER.info("Sending reload message for table: {} with forceDownload: {}, onlyIfNeeded: {}, and target: {}",
+ tableNameWithType, forceDownload, onlyIfNeeded, targetInstance == null ? "every instance" : targetInstance);
ClusterMessagingService messagingService = _helixZkManager.getMessagingService();
- SegmentReloadMessage message = new SegmentReloadMessage(tableNameWithType, forceDownload);
+ SegmentReloadMessage message =
+ new SegmentReloadMessage(tableNameWithType, null, forceDownload, onlyIfNeeded, null);
int numMessagesSent =
MessagingServiceUtils.send(messagingService, message, tableNameWithType, null, targetInstance);
if (numMessagesSent > 0) {
@@ -3239,11 +3254,20 @@ public class PinotHelixResourceManager {
public Pair<Integer, String> reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload,
@Nullable String targetInstance) {
- LOGGER.info("Sending reload message for segment: {} in table: {} with forceDownload: {}, and target: {}",
- segmentName, tableNameWithType, forceDownload, targetInstance == null ? "every instance" : targetInstance);
+ return reloadSegment(tableNameWithType, segmentName, forceDownload, false, targetInstance);
+ }
+
+ /// Sends a reload message for a single segment. When `onlyIfNeeded` is true, servers skip the reload if the loaded
+ /// copy is already up to date.
+ public Pair<Integer, String> reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload,
+ boolean onlyIfNeeded, @Nullable String targetInstance) {
+ LOGGER.info("Sending reload message for segment: {} in table: {} with forceDownload: {}, onlyIfNeeded: {}, and "
+ + "target: {}", segmentName, tableNameWithType, forceDownload, onlyIfNeeded,
+ targetInstance == null ? "every instance" : targetInstance);
ClusterMessagingService messagingService = _helixZkManager.getMessagingService();
- SegmentReloadMessage message = new SegmentReloadMessage(tableNameWithType, List.of(segmentName), forceDownload);
+ SegmentReloadMessage message =
+ new SegmentReloadMessage(tableNameWithType, List.of(segmentName), forceDownload, onlyIfNeeded, null);
int numMessagesSent =
MessagingServiceUtils.send(messagingService, message, tableNameWithType, segmentName, targetInstance);
if (numMessagesSent > 0) {
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadService.java b/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadService.java
index aa86d9eabd8..7a5dcc327e5 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadService.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/services/PinotTableReloadService.java
@@ -81,13 +81,14 @@ public class PinotTableReloadService {
}
public SuccessResponse reloadSegment(String tableName, String segmentName, boolean forceDownload,
- String targetInstance, HttpHeaders headers) {
+ boolean onlyIfNeeded, String targetInstance, HttpHeaders headers) {
tableName = DatabaseUtils.translateTableName(tableName, headers);
long startTimeMs = System.currentTimeMillis();
segmentName = URIUtils.decode(segmentName);
String tableNameWithType = getExistingTable(tableName, segmentName);
Pair<Integer, String> msgInfo =
- _pinotHelixResourceManager.reloadSegment(tableNameWithType, segmentName, forceDownload, targetInstance);
+ _pinotHelixResourceManager.reloadSegment(tableNameWithType, segmentName, forceDownload, onlyIfNeeded,
+ targetInstance);
boolean zkJobMetaWriteSuccess = false;
int numReloadMsgSent = msgInfo.getLeft();
if (numReloadMsgSent > 0) {
@@ -113,7 +114,7 @@ public class PinotTableReloadService {
}
public SuccessResponse reloadAllSegments(String tableName, @Nullable String tableTypeStr, boolean forceDownload,
- @Nullable String targetInstance, @Nullable String instanceToSegmentsMapInJson,
+ boolean onlyIfNeeded, @Nullable String targetInstance, @Nullable String instanceToSegmentsMapInJson,
@Nullable String startTimestampStr, @Nullable String endTimestampStr, boolean excludeOverlapping,
HttpHeaders headers)
throws IOException {
@@ -128,7 +129,7 @@ public class PinotTableReloadService {
Response.Status.BAD_REQUEST);
}
return reloadSegmentsInTimeRange(tableName, tableTypeStr, startTimestampStr, endTimestampStr,
- excludeOverlapping, forceDownload, null, headers);
+ excludeOverlapping, forceDownload, onlyIfNeeded, null, headers);
} else if (excludeOverlapping) {
throw new ControllerApplicationException(LOG,
"excludeOverlapping can only be used together with startTimestamp/endTimestamp.",
@@ -144,7 +145,7 @@ public class PinotTableReloadService {
boolean hasAnyMessages = false;
for (String tableNameWithType : tableNamesWithType) {
Map<String, Map<String, String>> instanceMsgData =
- reloadSegmentsForTable(tableNameWithType, forceDownload, instanceToSegmentsMap);
+ reloadSegmentsForTable(tableNameWithType, forceDownload, onlyIfNeeded, instanceToSegmentsMap);
tableInstanceMsgData.put(tableNameWithType, instanceMsgData);
if (!instanceMsgData.isEmpty()) {
hasAnyMessages = true;
@@ -161,7 +162,7 @@ public class PinotTableReloadService {
Map<String, Map<String, String>> perTableMsgData = new LinkedHashMap<>();
for (String tableNameWithType : tableNamesWithType) {
Pair<Integer, String> msgInfo =
- _pinotHelixResourceManager.reloadAllSegments(tableNameWithType, forceDownload, targetInstance);
+ _pinotHelixResourceManager.reloadAllSegments(tableNameWithType, forceDownload, onlyIfNeeded, targetInstance);
int numReloadMsgSent = msgInfo.getLeft();
if (numReloadMsgSent <= 0) {
continue;
@@ -194,7 +195,7 @@ public class PinotTableReloadService {
public SuccessResponse reloadSegmentsInTimeRange(String tableName, @Nullable String tableTypeStr,
@Nullable String startTimestampStr, @Nullable String endTimestampStr, boolean excludeOverlapping,
- boolean forceDownload, @Nullable String targetInstance, HttpHeaders headers) {
+ boolean forceDownload, boolean onlyIfNeeded, @Nullable String targetInstance, HttpHeaders headers) {
long startTimestamp = Long.MIN_VALUE;
long endTimestamp = Long.MAX_VALUE;
if (Strings.isNullOrEmpty(startTimestampStr) && Strings.isNullOrEmpty(endTimestampStr)) {
@@ -251,8 +252,8 @@ public class PinotTableReloadService {
String reloadJobId = UUID.randomUUID().toString();
long startTimeMs = System.currentTimeMillis();
Map<String, Integer> instanceMsgCountMap =
- _pinotHelixResourceManager.reloadSegments(tableNameWithType, forceDownload, filteredInstanceToSegmentsMap,
- reloadJobId);
+ _pinotHelixResourceManager.reloadSegments(tableNameWithType, forceDownload, onlyIfNeeded,
+ filteredInstanceToSegmentsMap, reloadJobId);
int numReloadMsgSent = 0;
for (int instanceMsgCount : instanceMsgCountMap.values()) {
numReloadMsgSent += instanceMsgCount;
@@ -362,14 +363,15 @@ public class PinotTableReloadService {
}
private Map<String, Map<String, String>> reloadSegmentsForTable(String tableNameWithType, boolean forceDownload,
- Map<String, List<String>> instanceToSegmentsMap) {
+ boolean onlyIfNeeded, Map<String, List<String>> instanceToSegmentsMap) {
long startTimeMs = System.currentTimeMillis();
// Use a single shared job ID across all instances so the logical reload operation can be tracked
// via a single ZK entry, consistent with the time-range reload path.
String reloadJobId = UUID.randomUUID().toString();
String targetInstance = instanceToSegmentsMap.size() == 1 ? instanceToSegmentsMap.keySet().iterator().next() : null;
Map<String, Integer> instanceMsgCountMap =
- _pinotHelixResourceManager.reloadSegments(tableNameWithType, forceDownload, instanceToSegmentsMap, reloadJobId);
+ _pinotHelixResourceManager.reloadSegments(tableNameWithType, forceDownload, onlyIfNeeded, instanceToSegmentsMap,
+ reloadJobId);
int totalNumReloadMsgSent = 0;
Map<String, Map<String, String>> instanceMsgData = new HashMap<>();
for (Map.Entry<String, Integer> instanceMsgInfo : instanceMsgCountMap.entrySet()) {
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/services/PinotTableReloadServiceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/services/PinotTableReloadServiceTest.java
index 32d1e21c27b..d793f61495a 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/services/PinotTableReloadServiceTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/services/PinotTableReloadServiceTest.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.api.exception.ControllerApplicationException;
@@ -75,19 +76,19 @@ public class PinotTableReloadServiceTest {
Map<String, Integer> reloadResult = new HashMap<>();
reloadResult.put("server1", 1);
reloadResult.put("server2", 1);
- when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), anyMap(), anyString()))
+ when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(false), anyMap(), anyString()))
.thenReturn(reloadResult);
when(helixResourceManager.addNewReloadSegmentJob(eq(tableNameWithType), anyString(), eq(null), anyString(),
anyLong(), anyInt())).thenReturn(true);
SuccessResponse response =
- service.reloadAllSegments(rawTableName, "OFFLINE", false, null, null, "1000", "2000", false, null);
+ service.reloadAllSegments(rawTableName, "OFFLINE", false, false, null, null, "1000", "2000", false, null);
@SuppressWarnings("unchecked")
ArgumentCaptor<Map<String, List<String>>> mapCaptor = ArgumentCaptor.forClass(Map.class);
ArgumentCaptor<String> jobIdCaptor = ArgumentCaptor.forClass(String.class);
verify(helixResourceManager)
- .reloadSegments(eq(tableNameWithType), eq(false), mapCaptor.capture(), jobIdCaptor.capture());
+ .reloadSegments(eq(tableNameWithType), eq(false), eq(false), mapCaptor.capture(), jobIdCaptor.capture());
Map<String, List<String>> capturedMap = mapCaptor.getValue();
assertEquals(capturedMap.get("server1"), List.of("seg_1"));
assertEquals(capturedMap.get("server2"), List.of("seg_2"));
@@ -126,17 +127,18 @@ public class PinotTableReloadServiceTest {
.thenReturn(List.of("seg_1"));
when(helixResourceManager.getServerToSegmentsMap(tableNameWithType, null, false))
.thenReturn(Map.of("server1", List.of("seg_1", "seg_2")));
- when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), anyMap(), anyString()))
+ when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(false), anyMap(), anyString()))
.thenReturn(Map.of("server1", 1));
when(helixResourceManager.addNewReloadSegmentJob(eq(tableNameWithType), anyString(), eq(null), anyString(),
anyLong(), anyInt())).thenReturn(true);
- service.reloadAllSegments("myTable", "OFFLINE", false, null, null, "1000", "2000", true, null);
+ service.reloadAllSegments("myTable", "OFFLINE", false, false, null, null, "1000", "2000", true, null);
verify(helixResourceManager).getSegmentsFor(tableNameWithType, true, 1000L, 2000L, true);
@SuppressWarnings("unchecked")
ArgumentCaptor<Map<String, List<String>>> mapCaptor = ArgumentCaptor.forClass(Map.class);
- verify(helixResourceManager).reloadSegments(eq(tableNameWithType), eq(false), mapCaptor.capture(), anyString());
+ verify(helixResourceManager)
+ .reloadSegments(eq(tableNameWithType), eq(false), eq(false), mapCaptor.capture(), anyString());
// Only seg_1 should be dispatched (seg_2 was excluded by the overlapping filter)
assertEquals(mapCaptor.getValue().get("server1"), List.of("seg_1"));
}
@@ -153,7 +155,7 @@ public class PinotTableReloadServiceTest {
ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments("rawTable", null, true, null, null, "0", "10", false, null));
+ () -> service.reloadAllSegments("rawTable", null, true, false, null, null, "0", "10", false, null));
assertNotNull(exception);
assertEquals(exception.getResponse().getStatus(), Response.Status.NOT_FOUND.getStatusCode());
verify(helixResourceManager).getExistingTableNamesWithType("rawTable", TableType.OFFLINE);
@@ -172,12 +174,12 @@ public class PinotTableReloadServiceTest {
.thenReturn(List.of("seg_1"));
when(helixResourceManager.getServerToSegmentsMap(tableNameWithType, null, false))
.thenReturn(Map.of("server1", List.of("seg_1")));
- when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), anyMap(), anyString()))
+ when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(false), anyMap(), anyString()))
.thenReturn(Map.of("server1", 1));
when(helixResourceManager.addNewReloadSegmentJob(eq(tableNameWithType), anyString(), eq("server1"), anyString(),
anyLong(), anyInt(), anyString())).thenReturn(true);
- service.reloadAllSegments("myTable", "OFFLINE", false, null, null, null, "2000", false, null);
+ service.reloadAllSegments("myTable", "OFFLINE", false, false, null, null, null, "2000", false, null);
verify(helixResourceManager).getSegmentsFor(tableNameWithType, true, Long.MIN_VALUE, 2000L, false);
}
@@ -195,12 +197,12 @@ public class PinotTableReloadServiceTest {
.thenReturn(List.of("seg_1"));
when(helixResourceManager.getServerToSegmentsMap(tableNameWithType, null, false))
.thenReturn(Map.of("server1", List.of("seg_1")));
- when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), anyMap(), anyString()))
+ when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(false), anyMap(), anyString()))
.thenReturn(Map.of("server1", 1));
when(helixResourceManager.addNewReloadSegmentJob(eq(tableNameWithType), anyString(), eq("server1"), anyString(),
anyLong(), anyInt(), anyString())).thenReturn(true);
- service.reloadAllSegments("myTable", "OFFLINE", false, null, null, "1000", null, false, null);
+ service.reloadAllSegments("myTable", "OFFLINE", false, false, null, null, "1000", null, false, null);
verify(helixResourceManager).getSegmentsFor(tableNameWithType, true, 1000L, Long.MAX_VALUE, false);
}
@@ -213,7 +215,8 @@ public class PinotTableReloadServiceTest {
ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments("myTable", "OFFLINE", false, null, null, "abc", "1000", false, null));
+ () -> service.reloadAllSegments("myTable", "OFFLINE", false, false, null, null, "abc", "1000", false,
+ null));
assertNotNull(exception);
assertEquals(exception.getResponse().getStatus(), Response.Status.BAD_REQUEST.getStatusCode());
verifyNoInteractions(helixResourceManager);
@@ -227,7 +230,8 @@ public class PinotTableReloadServiceTest {
ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments("myTable", "OFFLINE", false, "server1", null, "0", "10", false, null));
+ () -> service.reloadAllSegments("myTable", "OFFLINE", false, false, "server1", null, "0", "10", false,
+ null));
assertNotNull(exception);
assertEquals(exception.getResponse().getStatus(), Response.Status.BAD_REQUEST.getStatusCode());
verifyNoInteractions(helixResourceManager);
@@ -241,8 +245,8 @@ public class PinotTableReloadServiceTest {
ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments("myTable", "OFFLINE", false, null, "{\"server1\":[\"seg_1\"]}", "0", "10",
- false, null));
+ () -> service.reloadAllSegments("myTable", "OFFLINE", false, false, null, "{\"server1\":[\"seg_1\"]}",
+ "0", "10", false, null));
assertNotNull(exception);
assertEquals(exception.getResponse().getStatus(), Response.Status.BAD_REQUEST.getStatusCode());
verifyNoInteractions(helixResourceManager);
@@ -263,13 +267,13 @@ public class PinotTableReloadServiceTest {
when(helixResourceManager.getServerToSegmentsMap(tableNameWithType, null, false))
.thenReturn(Map.of("server1", List.of("seg_1")));
- when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), anyMap(), anyString()))
+ when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(false), anyMap(), anyString()))
.thenReturn(Map.of("server1", 1));
when(helixResourceManager.addNewReloadSegmentJob(eq(tableNameWithType), anyString(), eq(null), anyString(),
anyLong(), anyInt())).thenReturn(false);
SuccessResponse response =
- service.reloadAllSegments(rawTableName, "OFFLINE", false, null, null, "1000", "2000", false, null);
+ service.reloadAllSegments(rawTableName, "OFFLINE", false, false, null, null, "1000", "2000", false, null);
Map<String, Map<String, String>> payload =
JsonUtils.stringToObject(response.getStatus(), new TypeReference<>() {
@@ -292,12 +296,13 @@ public class PinotTableReloadServiceTest {
.thenReturn(List.of("seg_1"));
when(helixResourceManager.getServerToSegmentsMap(tableNameWithType, null, false))
.thenReturn(Map.of("server1", List.of("seg_1")));
- when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), anyMap(), anyString()))
+ when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(false), anyMap(), anyString()))
.thenReturn(Map.of("server1", 0));
ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments(rawTableName, "OFFLINE", false, null, null, "1000", "2000", false, null));
+ () -> service.reloadAllSegments(rawTableName, "OFFLINE", false, false, null, null, "1000", "2000", false,
+ null));
assertEquals(exception.getResponse().getStatus(), Response.Status.NOT_FOUND.getStatusCode());
}
@@ -315,15 +320,16 @@ public class PinotTableReloadServiceTest {
Map<String, List<String>> instanceToSegmentsMap = Map.of(
"server1", List.of("seg_1"),
"server2", List.of("seg_2"));
- when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(instanceToSegmentsMap), anyString()))
+ when(helixResourceManager.reloadSegments(eq(tableNameWithType), eq(false), eq(false), eq(instanceToSegmentsMap),
+ anyString()))
.thenReturn(Map.of(
"server1", 1,
"server2", 1));
when(helixResourceManager.addNewReloadSegmentJob(eq(tableNameWithType), anyString(), eq(null), anyString(),
anyLong(), anyInt(), anyString())).thenReturn(true);
- service.reloadAllSegments(rawTableName, "OFFLINE", false, null, JsonUtils.objectToString(instanceToSegmentsMap),
- null, null, false, null);
+ service.reloadAllSegments(rawTableName, "OFFLINE", false, false, null,
+ JsonUtils.objectToString(instanceToSegmentsMap), null, null, false, null);
ArgumentCaptor<String> instanceToSegmentsMapCaptor = ArgumentCaptor.forClass(String.class);
verify(helixResourceManager).addNewReloadSegmentJob(eq(tableNameWithType), anyString(), eq(null), anyString(),
@@ -343,17 +349,55 @@ public class PinotTableReloadServiceTest {
// startTimestamp == endTimestamp
ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments("myTable", "OFFLINE", false, null, null, "1000", "1000", false, null));
+ () -> service.reloadAllSegments("myTable", "OFFLINE", false, false, null, null, "1000", "1000", false,
+ null));
assertEquals(exception.getResponse().getStatus(), Response.Status.BAD_REQUEST.getStatusCode());
// startTimestamp > endTimestamp
exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments("myTable", "OFFLINE", false, null, null, "2000", "1000", false, null));
+ () -> service.reloadAllSegments("myTable", "OFFLINE", false, false, null, null, "2000", "1000", false,
+ null));
assertEquals(exception.getResponse().getStatus(), Response.Status.BAD_REQUEST.getStatusCode());
verifyNoInteractions(helixResourceManager);
}
+ @Test
+ public void testReloadOnlyIfNeededIsPassedToServers() throws Exception {
+ PinotHelixResourceManager helixResourceManager = mock(PinotHelixResourceManager.class);
+ PinotTableReloadService service = new PinotTableReloadService(helixResourceManager, new ControllerConf(),
+ mock(Executor.class), mock(HttpClientConnectionManager.class));
+ String tableNameWithType = "myTable_OFFLINE";
+ when(helixResourceManager.getExistingTableNamesWithType("myTable", TableType.OFFLINE))
+ .thenReturn(List.of(tableNameWithType));
+ when(helixResourceManager.getExistingTableNamesWithType(tableNameWithType, TableType.OFFLINE))
+ .thenReturn(List.of(tableNameWithType));
+ when(helixResourceManager.reloadAllSegments(tableNameWithType, false, true, null)).thenReturn(Pair.of(1, "job"));
+ when(helixResourceManager.reloadSegment(tableNameWithType, "seg_1", false, true, null))
+ .thenReturn(Pair.of(1, "job"));
+
+ service.reloadAllSegments("myTable", "OFFLINE", false, true, null, null, null, null, false, null);
+ verify(helixResourceManager).reloadAllSegments(tableNameWithType, false, true, null);
+
+ service.reloadSegment(tableNameWithType, "seg_1", false, true, null, null);
+ verify(helixResourceManager).reloadSegment(tableNameWithType, "seg_1", false, true, null);
+ }
+
+ @Test
+ public void testReloadAcceptsForceDownloadWithOnlyIfNeeded() throws Exception {
+ // Both flags are forwarded as-is; servers give forceDownload precedence
+ PinotHelixResourceManager helixResourceManager = mock(PinotHelixResourceManager.class);
+ PinotTableReloadService service = new PinotTableReloadService(helixResourceManager, new ControllerConf(),
+ mock(Executor.class), mock(HttpClientConnectionManager.class));
+ String tableNameWithType = "myTable_OFFLINE";
+ when(helixResourceManager.getExistingTableNamesWithType("myTable", TableType.OFFLINE))
+ .thenReturn(List.of(tableNameWithType));
+ when(helixResourceManager.reloadAllSegments(tableNameWithType, true, true, null)).thenReturn(Pair.of(1, "job"));
+
+ service.reloadAllSegments("myTable", "OFFLINE", true, true, null, null, null, null, false, null);
+ verify(helixResourceManager).reloadAllSegments(tableNameWithType, true, true, null);
+ }
+
@Test
public void testReloadAllSegmentsRejectsExcludeOverlappingWithoutRange() {
PinotHelixResourceManager helixResourceManager = mock(PinotHelixResourceManager.class);
@@ -362,7 +406,7 @@ public class PinotTableReloadServiceTest {
ControllerApplicationException exception =
expectThrows(ControllerApplicationException.class,
- () -> service.reloadAllSegments("myTable", "OFFLINE", false, null, null, null, null, true, null));
+ () -> service.reloadAllSegments("myTable", "OFFLINE", false, false, null, null, null, null, true, null));
assertEquals(exception.getResponse().getStatus(), Response.Status.BAD_REQUEST.getStatusCode());
verifyNoInteractions(helixResourceManager);
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
index fb83b87128b..28f36f7578b 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java
@@ -74,6 +74,7 @@ import org.apache.pinot.core.util.PeerServerSegmentFinder;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.StaleSegment;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.local.indexsegment.immutable.EmptyIndexSegment;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
@@ -712,15 +713,23 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
@Override
- public void reloadSegment(String segmentName, boolean forceDownload, String reloadJobId)
+ public void reloadSegment(String segmentName, boolean forceDownload, boolean onlyIfNeeded, String reloadJobId)
throws Exception {
Preconditions.checkState(!_shutDown,
"Table data manager is already shut down, cannot reload segment: %s in table: %s", segmentName,
_tableNameWithType);
- _logger.info("Reloading segment: {} with forceDownload: {}", segmentName, forceDownload);
+ _logger.info("Reloading segment: {} with forceDownload: {}, onlyIfNeeded: {}", segmentName, forceDownload,
+ onlyIfNeeded);
SegmentDataManager segmentDataManager = _segmentDataManagerMap.get(segmentName);
if (segmentDataManager != null) {
IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig();
+ if (onlyIfNeeded && !forceDownload && !checkReloadNeeded(segmentDataManager, indexLoadingConfig)) {
+ _logger.info("Skipped segment: {} with onlyIfNeeded as it is already up to date", segmentName);
+ if (reloadJobId != null) {
+ _reloadJobStatusCache.recordSkipped(reloadJobId, List.of(segmentName));
+ }
+ return;
+ }
reloadSegment(segmentDataManager, indexLoadingConfig, forceDownload);
} else {
_logger.warn("Failed to find segment: {}, skipping reloading it", segmentName);
@@ -728,25 +737,33 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
@Override
- public void reloadAllSegments(boolean forceDownload, String reloadJobId)
+ public void reloadAllSegments(boolean forceDownload, boolean onlyIfNeeded, String reloadJobId)
throws Exception {
Preconditions.checkState(!_shutDown,
"Table data manager is already shut down, cannot reload all segments in table: %s", _tableNameWithType);
- _logger.info("Reloading all segments with forceDownload: {}", forceDownload);
+ _logger.info("Reloading all segments with forceDownload: {}, onlyIfNeeded: {}", forceDownload, onlyIfNeeded);
List<SegmentDataManager> segmentDataManagers = new ArrayList<>(_segmentDataManagerMap.values());
if (!segmentDataManagers.isEmpty()) {
- reloadSegments(segmentDataManagers, fetchIndexLoadingConfig(), forceDownload, reloadJobId);
+ IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig();
+ List<SegmentDataManager> segmentsToReload = onlyIfNeeded && !forceDownload
+ ? selectSegmentsNeedingReload(segmentDataManagers, indexLoadingConfig, reloadJobId) : segmentDataManagers;
+ if (!segmentsToReload.isEmpty()) {
+ reloadSegments(segmentsToReload, indexLoadingConfig, forceDownload, reloadJobId);
+ }
}
- _logger.info("Reloaded all {} segments with forceDownload: {}", segmentDataManagers.size(), forceDownload);
+ _logger.info("Reloaded all {} segments with forceDownload: {}, onlyIfNeeded: {}", segmentDataManagers.size(),
+ forceDownload, onlyIfNeeded);
}
@Override
- public void reloadSegments(List<String> segmentNames, boolean forceDownload, String reloadJobId)
+ public void reloadSegments(List<String> segmentNames, boolean forceDownload, boolean onlyIfNeeded,
+ String reloadJobId)
throws Exception {
Preconditions.checkState(!_shutDown,
"Table data manager is already shut down, cannot reload segments: %s in table: %s", segmentNames,
_tableNameWithType);
- _logger.info("Reloading segments: {} with forceDownload: {}", segmentNames, forceDownload);
+ _logger.info("Reloading segments: {} with forceDownload: {}, onlyIfNeeded: {}", segmentNames, forceDownload,
+ onlyIfNeeded);
List<SegmentDataManager> segmentDataManagers = new ArrayList<>(segmentNames.size());
List<String> missingSegments = new ArrayList<>();
for (String segmentName : segmentNames) {
@@ -758,13 +775,19 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
}
if (!segmentDataManagers.isEmpty()) {
- reloadSegments(segmentDataManagers, fetchIndexLoadingConfig(), forceDownload, reloadJobId);
+ IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig();
+ List<SegmentDataManager> segmentsToReload = onlyIfNeeded && !forceDownload
+ ? selectSegmentsNeedingReload(segmentDataManagers, indexLoadingConfig, reloadJobId) : segmentDataManagers;
+ if (!segmentsToReload.isEmpty()) {
+ reloadSegments(segmentsToReload, indexLoadingConfig, forceDownload, reloadJobId);
+ }
}
if (missingSegments.isEmpty()) {
- _logger.info("Reloaded segments: {} with forceDownload: {}", segmentNames, forceDownload);
+ _logger.info("Reloaded {} segments with forceDownload: {}, onlyIfNeeded: {}", segmentNames.size(), forceDownload,
+ onlyIfNeeded);
} else {
- _logger.warn("Reloaded {}/{} segments: {} with forceDownload: {}, missing segments: {}",
- segmentDataManagers.size(), segmentNames.size(), segmentNames, forceDownload, missingSegments);
+ _logger.warn("Reloaded {}/{} segments with forceDownload: {}, onlyIfNeeded: {}, missing segments: {}",
+ segmentDataManagers.size(), segmentNames.size(), forceDownload, onlyIfNeeded, missingSegments);
}
}
@@ -1085,6 +1108,104 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
}
+ /// Returns the subset of `segmentDataManagers` that [#isReloadNeeded] reports as out of date, evaluating the check
+ /// in parallel on the reload executor. Skipped segments are counted in the reload job status (as successful) when
+ /// `reloadJobId` is set. A segment is kept (i.e. reloaded as usual) when its check fails, so an error never turns
+ /// into a silently skipped reload.
+ ///
+ /// The check runs against the currently loaded copy without holding the segment lock. Skipping on that snapshot is
+ /// safe: anything that later changes the segment (a replace or another reload) loads it against the then-current
+ /// config, and a segment found to need reload still goes through the regular reload path, which re-evaluates
+ /// everything under the lock.
+ @VisibleForTesting
+ List<SegmentDataManager> selectSegmentsNeedingReload(List<SegmentDataManager> segmentDataManagers,
+ IndexLoadingConfig indexLoadingConfig, @Nullable String reloadJobId)
+ throws Exception {
+ boolean[] needReload = new boolean[segmentDataManagers.size()];
+ CompletableFuture<?>[] futures = new CompletableFuture<?>[segmentDataManagers.size()];
+ for (int i = 0; i < segmentDataManagers.size(); i++) {
+ int index = i;
+ SegmentDataManager segmentDataManager = segmentDataManagers.get(i);
+ futures[i] = CompletableFuture.runAsync(
+ () -> needReload[index] = checkReloadNeeded(segmentDataManager, indexLoadingConfig), _segmentReloadExecutor);
+ }
+ CompletableFuture.allOf(futures).get();
+
+ List<SegmentDataManager> segmentsToReload = new ArrayList<>();
+ List<String> skippedSegments = new ArrayList<>();
+ for (int i = 0; i < segmentDataManagers.size(); i++) {
+ if (needReload[i]) {
+ segmentsToReload.add(segmentDataManagers.get(i));
+ } else {
+ skippedSegments.add(segmentDataManagers.get(i).getSegmentName());
+ }
+ }
+ if (!skippedSegments.isEmpty()) {
+ _logger.info("Skipping reload for {}/{} segments with onlyIfNeeded as they are already up to date",
+ skippedSegments.size(), segmentDataManagers.size());
+ if (reloadJobId != null) {
+ _reloadJobStatusCache.recordSkipped(reloadJobId, skippedSegments);
+ }
+ }
+ return segmentsToReload;
+ }
+
+ /// Runs [#isReloadNeeded] while holding a reference, so the segment cannot be destroyed while it is inspected.
+ /// Returns `true` (i.e. reload as usual) when the segment is already being destroyed, as it was replaced or offloaded
+ /// concurrently and the regular reload path handles that, or when the check fails, so an error never turns into a
+ /// silently skipped reload.
+ private boolean checkReloadNeeded(SegmentDataManager segmentDataManager, IndexLoadingConfig indexLoadingConfig) {
+ if (!segmentDataManager.increaseReferenceCount()) {
+ return true;
+ }
+ try {
+ return isReloadNeeded(segmentDataManager, indexLoadingConfig);
+ } catch (Exception e) {
+ _logger.warn("Caught exception while checking whether segment: {} needs reload, reloading it",
+ segmentDataManager.getSegmentName(), e);
+ return true;
+ } finally {
+ releaseSegment(segmentDataManager);
+ }
+ }
+
+ /// Returns whether reloading the segment would change what is loaded, i.e. whether the regular reload path would
+ /// download a new copy (completed segment whose CRC differs from ZK), move it to a different tier, or preprocess it
+ /// under the given index loading config. Consuming segments are never considered as needing reload, as reloading
+ /// them means force committing. Segment types the check does not understand are considered as needing reload.
+ protected boolean isReloadNeeded(SegmentDataManager segmentDataManager, IndexLoadingConfig indexLoadingConfig)
+ throws Exception {
+ if (segmentDataManager instanceof RealtimeSegmentDataManager) {
+ return false;
+ }
+ IndexSegment segment = segmentDataManager.getSegment();
+ if (!(segment instanceof ImmutableSegmentImpl) && !(segment instanceof EmptyIndexSegment)) {
+ return true;
+ }
+ String segmentName = segmentDataManager.getSegmentName();
+ SegmentZKMetadata zkMetadata = fetchZKMetadata(segmentName);
+ if (shouldDownloadForCrcMismatch(zkMetadata, segment.getSegmentMetadata())) {
+ return true;
+ }
+ String segmentTier = getSegmentCurrentTier(segmentName);
+ IndexLoadingConfig segmentIndexLoadingConfig = indexLoadingConfig.copyWithSegmentTier(segmentTier);
+ SegmentDirectoryLoader segmentDirectoryLoader =
+ SegmentDirectoryLoaderRegistry.getSegmentDirectoryLoader(segmentIndexLoadingConfig.getSegmentDirectoryLoader());
+ if (segmentDirectoryLoader.needsTierMigration(zkMetadata.getTier(), segmentTier)) {
+ return true;
+ }
+ // An empty segment has no indexes to preprocess: reloading it loads another empty segment
+ return segment instanceof ImmutableSegmentImpl
+ && ((ImmutableSegmentImpl) segment).isReloadNeeded(segmentIndexLoadingConfig);
+ }
+
+ /// Whether a completed segment has to be re-downloaded because its local CRC no longer matches ZK. The completed
+ /// status confirms the new copy is available in deep store or with a peer before the local copy is discarded.
+ protected boolean shouldDownloadForCrcMismatch(SegmentZKMetadata zkMetadata, SegmentMetadata localMetadata) {
+ return isSegmentStatusCompleted(zkMetadata) && !hasSameCRC(zkMetadata, localMetadata)
+ && _instanceDataManagerConfig.shouldCheckCRCOnSegmentLoad();
+ }
+
protected void reloadSegments(List<SegmentDataManager> segmentDataManagers, IndexLoadingConfig indexLoadingConfig,
boolean forceDownload, String reloadJobId)
throws Exception {
@@ -1188,9 +1309,7 @@ public abstract class BaseTableDataManager implements TableDataManager {
- Continue loading the segment from the index directory.
*/
boolean crcMatch = hasSameCRC(zkMetadata, localMetadata);
- boolean shouldDownload = forceDownload
- || (isSegmentStatusCompleted(zkMetadata) && !crcMatch
- && _instanceDataManagerConfig.shouldCheckCRCOnSegmentLoad());
+ boolean shouldDownload = forceDownload || shouldDownloadForCrcMismatch(zkMetadata, localMetadata);
// For an upsert table with metadata TTL and snapshots enabled, a reload must not blindly re-scan
// the segment and re-add every primary key, as that would resurrect keys already expired (metadataTTL)
// or deleted (deletedKeysTTL). Instead we rebuild the upsert metadata from the persisted validDocIds snapshot
@@ -1726,6 +1845,9 @@ public abstract class BaseTableDataManager implements TableDataManager {
}
}
+ /// Uses the same per-segment check as an only-if-needed reload ([#isReloadNeeded]), so this reports `true` exactly
+ /// when such a reload would reload at least one segment. A segment whose check fails is reported as needing reload,
+ /// matching how the reload treats it.
@Override
public boolean needReloadSegments()
throws Exception {
@@ -1734,14 +1856,15 @@ public abstract class BaseTableDataManager implements TableDataManager {
boolean needReload = false;
try {
for (SegmentDataManager segmentDataManager : segmentDataManagers) {
- IndexSegment segment = segmentDataManager.getSegment();
- if (segment instanceof ImmutableSegmentImpl) {
- ImmutableSegmentImpl immutableSegment = (ImmutableSegmentImpl) segment;
- indexLoadingConfig.setSegmentTier(immutableSegment.getTier());
- if (immutableSegment.isReloadNeeded(indexLoadingConfig)) {
- needReload = true;
- break;
- }
+ try {
+ needReload = isReloadNeeded(segmentDataManager, indexLoadingConfig);
+ } catch (Exception e) {
+ _logger.warn("Caught exception while checking whether segment: {} needs reload",
+ segmentDataManager.getSegmentName(), e);
+ needReload = true;
+ }
+ if (needReload) {
+ break;
}
}
} finally {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java
index 7e8647784bd..71d384fc6b5 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/InstanceDataManager.java
@@ -101,15 +101,38 @@ public interface InstanceDataManager {
/// Reloads a segment in a table. This method can download a new segment to replace the local one before loading.
/// Download happens when local segment's CRC mismatches the one of the remote segment; but can also be forced to do
/// regardless of CRC.
- void reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload, String reloadJobId)
+ default void reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload, String reloadJobId)
+ throws Exception {
+ reloadSegment(tableNameWithType, segmentName, forceDownload, false, reloadJobId);
+ }
+
+ /// Same as [#reloadSegment(String, String, boolean, String)], but when `onlyIfNeeded` is true (and `forceDownload`
+ /// is false) the segment is skipped if its loaded copy is already up to date. See
+ /// [TableDataManager#reloadSegment(String, boolean, boolean, String)].
+ void reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload, boolean onlyIfNeeded,
+ String reloadJobId)
throws Exception;
/// Reloads all segments of a table.
- void reloadAllSegments(String tableNameWithType, boolean forceDownload, String reloadJobId)
+ default void reloadAllSegments(String tableNameWithType, boolean forceDownload, String reloadJobId)
+ throws Exception {
+ reloadAllSegments(tableNameWithType, forceDownload, false, reloadJobId);
+ }
+
+ /// Reloads all segments of a table, skipping up-to-date segments when `onlyIfNeeded` is true.
+ void reloadAllSegments(String tableNameWithType, boolean forceDownload, boolean onlyIfNeeded, String reloadJobId)
throws Exception;
/// Reload a list of segments in a table.
- void reloadSegments(String tableNameWithType, List<String> segmentNames, boolean forceDownload, String reloadJobId)
+ default void reloadSegments(String tableNameWithType, List<String> segmentNames, boolean forceDownload,
+ String reloadJobId)
+ throws Exception {
+ reloadSegments(tableNameWithType, segmentNames, forceDownload, false, reloadJobId);
+ }
+
+ /// Reload a list of segments in a table, skipping up-to-date segments when `onlyIfNeeded` is true.
+ void reloadSegments(String tableNameWithType, List<String> segmentNames, boolean forceDownload,
+ boolean onlyIfNeeded, String reloadJobId)
throws Exception;
/// Returns all tables served by the instance.
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
index 9948fca02fe..1bfadf932d0 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/BaseTableDataManagerTest.java
@@ -53,8 +53,10 @@ import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
import org.apache.pinot.common.utils.helix.FakePropertyStore;
import org.apache.pinot.core.data.manager.offline.ImmutableSegmentDataManager;
import org.apache.pinot.core.data.manager.offline.OfflineTableDataManager;
+import org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.indexsegment.immutable.EmptyIndexSegment;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
@@ -98,10 +100,12 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -627,6 +631,194 @@ public class BaseTableDataManagerTest {
}
}
+ @Test
+ public void testReloadOnlyIfNeededSkipsUpToDateSegment()
+ throws Exception {
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ SegmentZKMetadata zkMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ zkMetadata.setCrc(getCRC(indexDir));
+
+ BaseTableDataManager tableDataManager = spy(createTableManager());
+ ImmutableSegmentDataManager segmentDataManager = loadSegmentDataManager(indexDir);
+ tableDataManager.registerSegment(SEGMENT_NAME, segmentDataManager);
+ seedZKMetadata(tableDataManager, SEGMENT_NAME, zkMetadata);
+ doAnswer(invocation -> new IndexLoadingConfig(DEFAULT_TABLE_CONFIG, SCHEMA)).when(tableDataManager)
+ .fetchIndexLoadingConfig();
+
+ tableDataManager.reloadSegments(List.of(SEGMENT_NAME), false, true, "job");
+ assertSame(tableDataManager.getSegmentDataManager(SEGMENT_NAME), segmentDataManager);
+ verify(tableDataManager, never()).reloadSegment(any(SegmentDataManager.class), any(IndexLoadingConfig.class),
+ anyBoolean());
+ verify(tableDataManager._reloadJobStatusCache).recordSkipped("job", List.of(SEGMENT_NAME));
+
+ tableDataManager.reloadSegment(SEGMENT_NAME, false, true, "job2");
+ assertSame(tableDataManager.getSegmentDataManager(SEGMENT_NAME), segmentDataManager);
+ verify(tableDataManager._reloadJobStatusCache).recordSkipped("job2", List.of(SEGMENT_NAME));
+
+ // Without onlyIfNeeded the same segment goes through the regular reload path. Use the single-segment API, which
+ // reloads synchronously in every implementation (batch reloads may complete asynchronously in subclasses).
+ tableDataManager.reloadSegment(SEGMENT_NAME, false, false, "job3");
+ assertNotSame(tableDataManager.getSegmentDataManager(SEGMENT_NAME), segmentDataManager);
+ }
+
+ @Test
+ public void testReloadOnlyIfNeededReloadsSegmentNeedingPreprocess()
+ throws Exception {
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ assertFalse(hasInvertedIndex(indexDir, STRING_COLUMN));
+ SegmentZKMetadata zkMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ zkMetadata.setCrc(getCRC(indexDir));
+
+ BaseTableDataManager tableDataManager = spy(createTableManager());
+ ImmutableSegmentDataManager segmentDataManager = loadSegmentDataManager(indexDir);
+ tableDataManager.registerSegment(SEGMENT_NAME, segmentDataManager);
+ seedZKMetadata(tableDataManager, SEGMENT_NAME, zkMetadata);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setInvertedIndexColumns(List.of(STRING_COLUMN)).build();
+ doAnswer(invocation -> new IndexLoadingConfig(tableConfig, SCHEMA)).when(tableDataManager)
+ .fetchIndexLoadingConfig();
+
+ // Single-segment API so the reload has completed when it returns (see testReloadOnlyIfNeededSkipsUpToDateSegment)
+ tableDataManager.reloadSegment(SEGMENT_NAME, false, true, "job");
+ assertNotSame(tableDataManager.getSegmentDataManager(SEGMENT_NAME), segmentDataManager);
+ assertTrue(hasInvertedIndex(indexDir, STRING_COLUMN));
+ verify(tableDataManager._reloadJobStatusCache, never()).recordSkipped(any(), any());
+ }
+
+ @Test
+ public void testReloadForceDownloadTakesPrecedenceOverOnlyIfNeeded()
+ throws Exception {
+ // The segment is up to date, so onlyIfNeeded alone would skip it; forceDownload must still re-download it
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ SegmentZKMetadata zkMetadata =
+ makeRawSegment(indexDir, new File(TEMP_DIR, SEGMENT_NAME + TarCompressionUtils.TAR_COMPRESSED_FILE_EXTENSION),
+ false);
+
+ BaseTableDataManager tableDataManager = spy(createTableManager());
+ ImmutableSegmentDataManager segmentDataManager = loadSegmentDataManager(indexDir);
+ tableDataManager.registerSegment(SEGMENT_NAME, segmentDataManager);
+ seedZKMetadata(tableDataManager, SEGMENT_NAME, zkMetadata);
+ doAnswer(invocation -> new IndexLoadingConfig(DEFAULT_TABLE_CONFIG, SCHEMA)).when(tableDataManager)
+ .fetchIndexLoadingConfig();
+
+ tableDataManager.reloadSegment(SEGMENT_NAME, true, true, "job");
+ assertNotSame(tableDataManager.getSegmentDataManager(SEGMENT_NAME), segmentDataManager);
+ verify(tableDataManager._reloadJobStatusCache, never()).recordSkipped(any(), any());
+ }
+
+ @Test
+ public void testIsReloadNeeded()
+ throws Exception {
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
+ BaseTableDataManager tableDataManager = spy(createTableManager());
+ ImmutableSegmentDataManager segmentDataManager = loadSegmentDataManager(indexDir);
+ tableDataManager.registerSegment(SEGMENT_NAME, segmentDataManager);
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(DEFAULT_TABLE_CONFIG, SCHEMA);
+
+ // Up to date
+ SegmentZKMetadata zkMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ zkMetadata.setCrc(crc);
+ seedZKMetadata(tableDataManager, SEGMENT_NAME, zkMetadata);
+ assertFalse(tableDataManager.isReloadNeeded(segmentDataManager, indexLoadingConfig));
+
+ // CRC changed on a completed segment, so the reload path would download it
+ zkMetadata.setCrc(crc + 1);
+ assertTrue(tableDataManager.isReloadNeeded(segmentDataManager, indexLoadingConfig));
+
+ // CRC changed but the segment is not completed yet, so the reload path would keep the local copy
+ zkMetadata.setStatus(CommonConstants.Segment.Realtime.Status.IN_PROGRESS);
+ assertFalse(tableDataManager.isReloadNeeded(segmentDataManager, indexLoadingConfig));
+
+ // Target tier differs from the current one
+ SegmentZKMetadata tierZkMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ tierZkMetadata.setCrc(crc);
+ tierZkMetadata.setTier(TIER_NAME);
+ seedZKMetadata(tableDataManager, SEGMENT_NAME, tierZkMetadata);
+ assertTrue(tableDataManager.isReloadNeeded(segmentDataManager,
+ createTierIndexLoadingConfig(TIER_TABLE_CONFIG)));
+
+ // Consuming segments are never reloaded in this mode, and unknown segment types always are
+ assertFalse(tableDataManager.isReloadNeeded(mock(RealtimeSegmentDataManager.class), indexLoadingConfig));
+ assertTrue(tableDataManager.isReloadNeeded(createImmutableSegmentDataManager(SEGMENT_NAME, crc),
+ indexLoadingConfig));
+ }
+
+ @Test
+ public void testIsReloadNeededForEmptySegment()
+ throws Exception {
+ File indexDir = createSegment(SegmentVersion.v3, 0);
+ long crc = getCRC(indexDir);
+ BaseTableDataManager tableDataManager = spy(createTableManager());
+ ImmutableSegmentDataManager segmentDataManager = loadSegmentDataManager(indexDir);
+ assertTrue(segmentDataManager.getSegment() instanceof EmptyIndexSegment);
+ tableDataManager.registerSegment(SEGMENT_NAME, segmentDataManager);
+ SegmentZKMetadata zkMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ zkMetadata.setCrc(crc);
+ seedZKMetadata(tableDataManager, SEGMENT_NAME, zkMetadata);
+
+ // A new index does not apply to an empty segment, so a reload would not change it
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setInvertedIndexColumns(List.of(STRING_COLUMN)).build();
+ assertFalse(tableDataManager.isReloadNeeded(segmentDataManager, new IndexLoadingConfig(tableConfig, SCHEMA)));
+
+ // A changed CRC still needs a download
+ zkMetadata.setCrc(crc + 1);
+ assertTrue(tableDataManager.isReloadNeeded(segmentDataManager, new IndexLoadingConfig(tableConfig, SCHEMA)));
+ }
+
+ @Test
+ public void testNeedReloadSegmentsUsesReloadCheck()
+ throws Exception {
+ File indexDir = createSegment(SegmentVersion.v3, 5);
+ long crc = getCRC(indexDir);
+ BaseTableDataManager tableDataManager = spy(createTableManager());
+ tableDataManager.registerSegment(SEGMENT_NAME, loadSegmentDataManager(indexDir));
+ SegmentZKMetadata zkMetadata = new SegmentZKMetadata(SEGMENT_NAME);
+ zkMetadata.setCrc(crc);
+ seedZKMetadata(tableDataManager, SEGMENT_NAME, zkMetadata);
+ doAnswer(invocation -> new IndexLoadingConfig(DEFAULT_TABLE_CONFIG, SCHEMA)).when(tableDataManager)
+ .fetchIndexLoadingConfig();
+ assertFalse(tableDataManager.needReloadSegments());
+
+ // A CRC change is reported, which the preprocessing-only check used to miss
+ zkMetadata.setCrc(crc + 1);
+ assertTrue(tableDataManager.needReloadSegments());
+
+ // A segment whose check fails is reported as needing reload
+ zkMetadata.setCrc(crc);
+ assertFalse(tableDataManager.needReloadSegments());
+ doThrow(new RuntimeException("injected")).when(tableDataManager).isReloadNeeded(any(), any());
+ assertTrue(tableDataManager.needReloadSegments());
+ }
+
+ @Test
+ public void testSelectSegmentsNeedingReloadKeepsSegmentsThatCannotBeChecked()
+ throws Exception {
+ BaseTableDataManager tableDataManager = spy(createTableManager());
+ IndexLoadingConfig indexLoadingConfig = new IndexLoadingConfig(DEFAULT_TABLE_CONFIG, SCHEMA);
+
+ SegmentDataManager failingCheck = mock(SegmentDataManager.class);
+ when(failingCheck.getSegmentName()).thenReturn("failingCheck");
+ when(failingCheck.increaseReferenceCount()).thenReturn(true);
+ doThrow(new RuntimeException("injected")).when(tableDataManager).isReloadNeeded(eq(failingCheck), any());
+
+ SegmentDataManager destroyed = mock(SegmentDataManager.class);
+ when(destroyed.getSegmentName()).thenReturn("destroyed");
+ when(destroyed.increaseReferenceCount()).thenReturn(false);
+
+ SegmentDataManager upToDate = mock(SegmentDataManager.class);
+ when(upToDate.getSegmentName()).thenReturn("upToDate");
+ when(upToDate.increaseReferenceCount()).thenReturn(true);
+ doReturn(false).when(tableDataManager).isReloadNeeded(eq(upToDate), any());
+
+ List<SegmentDataManager> selected = tableDataManager.selectSegmentsNeedingReload(
+ List.of(failingCheck, destroyed, upToDate), indexLoadingConfig, "job");
+ assertEquals(selected, List.of(failingCheck, destroyed));
+ verify(tableDataManager, never()).isReloadNeeded(eq(destroyed), any());
+ verify(tableDataManager._reloadJobStatusCache).recordSkipped("job", List.of("upToDate"));
+ }
+
@Test
public void testReplaceSegmentNewData()
throws Exception {
@@ -1591,6 +1783,12 @@ public class BaseTableDataManagerTest {
return indexLoadingConfig;
}
+ private static ImmutableSegmentDataManager loadSegmentDataManager(File indexDir)
+ throws Exception {
+ return new ImmutableSegmentDataManager(
+ ImmutableSegmentLoader.load(indexDir, new IndexLoadingConfig(DEFAULT_TABLE_CONFIG, SCHEMA)));
+ }
+
protected ImmutableSegmentDataManager createImmutableSegmentDataManager(String segmentName, long crc) {
SegmentMetadata segmentMetadata = mock(SegmentMetadata.class);
when(segmentMetadata.getCrc()).thenReturn(crc);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java
index 8e90814fb8d..3ac165a8075 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/data/manager/TableDataManager.java
@@ -169,15 +169,41 @@ public interface TableDataManager {
/// `forceDownload` flag is true.
/// This operation is conducted within a failure handling framework and made transparent to ongoing queries, because
/// the segment is in online serving state.
- void reloadSegment(String segmentName, boolean forceDownload, String reloadJobId)
+ default void reloadSegment(String segmentName, boolean forceDownload, String reloadJobId)
+ throws Exception {
+ reloadSegment(segmentName, forceDownload, false, reloadJobId);
+ }
+
+ /// Same as [#reloadSegment(String, boolean, String)], but when `onlyIfNeeded` is true (and `forceDownload` is false)
+ /// the segment is skipped if its loaded copy is already up to date: same CRC as ZK, already on the target tier, and
+ /// no preprocessing needed under the current table config and schema. Consuming segments are always skipped in this
+ /// mode. Skipped segments are recorded in the reload job status and reported as successful.
+ ///
+ /// This does not cover settings that only take effect when a segment is loaded (e.g. read mode or tier storage
+ /// properties) or rebuilding derived state such as upsert metadata; a full reload is still needed for those.
+ void reloadSegment(String segmentName, boolean forceDownload, boolean onlyIfNeeded, String reloadJobId)
throws Exception;
/// Reloads all segments.
- void reloadAllSegments(boolean forceDownload, String reloadJobId)
+ default void reloadAllSegments(boolean forceDownload, String reloadJobId)
+ throws Exception {
+ reloadAllSegments(forceDownload, false, reloadJobId);
+ }
+
+ /// Reloads all segments, skipping up-to-date ones when `onlyIfNeeded` is true. See
+ /// [#reloadSegment(String, boolean, boolean, String)].
+ void reloadAllSegments(boolean forceDownload, boolean onlyIfNeeded, String reloadJobId)
throws Exception;
/// Reloads a list of segments.
- void reloadSegments(List<String> segmentNames, boolean forceDownload, String reloadJobId)
+ default void reloadSegments(List<String> segmentNames, boolean forceDownload, String reloadJobId)
+ throws Exception {
+ reloadSegments(segmentNames, forceDownload, false, reloadJobId);
+ }
+
+ /// Reloads a list of segments, skipping up-to-date ones when `onlyIfNeeded` is true. See
+ /// [#reloadSegment(String, boolean, boolean, String)].
+ void reloadSegments(List<String> segmentNames, boolean forceDownload, boolean onlyIfNeeded, String reloadJobId)
throws Exception;
/// Get the segment data directory, considering the segment tier if provided.
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ReloadJobStatus.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ReloadJobStatus.java
index 25874f055ca..7b85eab38d2 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ReloadJobStatus.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ReloadJobStatus.java
@@ -19,8 +19,11 @@
package org.apache.pinot.segment.local.utils;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.pinot.common.response.server.SegmentReloadFailureResponse;
@@ -30,6 +33,10 @@ import org.apache.pinot.common.response.server.SegmentReloadFailureResponse;
public class ReloadJobStatus {
private final String _jobId;
private final AtomicInteger _failureCount = new AtomicInteger(0);
+ // Names of the segments an only-if-needed reload left in place because their loaded copy was already up to date.
+ // Kept by name (rather than as a count) so the status check can tell a skipped segment apart from one that a later
+ // operation reloaded, and so a skipped segment that was since dropped is not counted.
+ private final Set<String> _skippedSegments = ConcurrentHashMap.newKeySet();
private final long _createdTimeMs;
private final List<SegmentReloadFailureResponse> _failedSegmentDetails = new ArrayList<>();
@@ -50,6 +57,18 @@ public class ReloadJobStatus {
return _failureCount.incrementAndGet();
}
+ public boolean hasSkippedSegments() {
+ return !_skippedSegments.isEmpty();
+ }
+
+ public boolean isSkipped(String segmentName) {
+ return _skippedSegments.contains(segmentName);
+ }
+
+ public void addSkippedSegments(Collection<String> segmentNames) {
+ _skippedSegments.addAll(segmentNames);
+ }
+
public long getCreatedTimeMs() {
return _createdTimeMs;
}
@@ -65,6 +84,7 @@ public class ReloadJobStatus {
@Override
public String toString() {
return "ReloadJobStatus{jobId='" + _jobId + "', failureCount=" + _failureCount.get()
+ + ", skippedCount=" + _skippedSegments.size()
+ ", failedSegmentDetailsCount=" + _failedSegmentDetails.size()
+ ", createdTimeMs=" + _createdTimeMs + '}';
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCache.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCache.java
index 76f7cff0080..193fc5b2dea 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCache.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCache.java
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
+import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -133,6 +134,17 @@ public class ServerReloadJobStatusCache implements PinotClusterConfigChangeListe
}
}
+ /// Records segments that an only-if-needed reload skipped because their loaded copy was already up to date.
+ ///
+ /// @param jobId reload job ID (UUID)
+ /// @param skippedSegments names of the skipped segments
+ public void recordSkipped(String jobId, Collection<String> skippedSegments) {
+ requireNonNull(jobId, "jobId cannot be null");
+ if (!skippedSegments.isEmpty()) {
+ getOrCreate(jobId).addSkippedSegments(skippedSegments);
+ }
+ }
+
/// Rebuilds the cache with new configuration and migrates existing entries.
/// This method is synchronized to prevent concurrent rebuilds.
///
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCacheTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCacheTest.java
index 09ef4f22263..fbc5844616a 100644
--- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCacheTest.java
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ServerReloadJobStatusCacheTest.java
@@ -279,6 +279,26 @@ public class ServerReloadJobStatusCacheTest {
assertThat(status.getFailedSegmentDetails().get(0).getSegmentName()).isEqualTo(segmentName);
}
+ @Test
+ public void testRecordSkippedAccumulates() {
+ ServerReloadJobStatusCache cache = new ServerReloadJobStatusCache("testServer");
+
+ cache.recordSkipped("job-skip", List.of());
+ assertThat(cache.getJobStatus("job-skip")).isNull();
+
+ cache.recordSkipped("job-skip", List.of("segment_1", "segment_2"));
+ cache.recordSkipped("job-skip", List.of("segment_2", "segment_3"));
+ cache.recordFailure("job-skip", "segment_4", new IOException("Test error"));
+ ReloadJobStatus status = cache.getJobStatus("job-skip");
+ assertThat(status).isNotNull();
+ assertThat(status.hasSkippedSegments()).isTrue();
+ assertThat(status.isSkipped("segment_1")).isTrue();
+ assertThat(status.isSkipped("segment_2")).isTrue();
+ assertThat(status.isSkipped("segment_3")).isTrue();
+ assertThat(status.isSkipped("segment_4")).isFalse();
+ assertThat(status.getFailureCount()).isEqualTo(1);
+ }
+
@org.testng.annotations.DataProvider(name = "failureLimits")
public Object[][] failureLimitsProvider() {
return new Object[][] {
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ControllerJobStatusResource.java b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ControllerJobStatusResource.java
index fc1cf1da447..fac41a41850 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ControllerJobStatusResource.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/ControllerJobStatusResource.java
@@ -23,7 +23,6 @@ import io.swagger.annotations.ApiOperation;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Optional;
import javax.inject.Inject;
import javax.inject.Singleton;
import javax.ws.rs.GET;
@@ -39,6 +38,7 @@ import org.apache.pinot.common.response.server.ServerReloadStatusResponse;
import org.apache.pinot.common.utils.DatabaseUtils;
import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.local.utils.ReloadJobStatus;
import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache;
import org.apache.pinot.segment.spi.creator.name.SegmentNameUtils;
import org.apache.pinot.server.starter.ServerInstance;
@@ -85,9 +85,15 @@ public class ControllerJobStatusResource {
totalSegmentCount = targetSegments.size();
}
try {
+ ReloadJobStatus jobStatus = reloadJobId != null ? _serverReloadJobStatusCache.getJobStatus(reloadJobId) : null;
+ // Segments skipped by an only-if-needed reload keep their old load time but are up to date, so they count as
+ // successful. Checking by name counts each segment once, even if it was reloaded again after being skipped, and
+ // leaves out skipped segments that have since been dropped.
+ boolean hasSkippedSegments = jobStatus != null && jobStatus.hasSkippedSegments();
int successCount = 0;
for (SegmentDataManager segmentDataManager : segmentDataManagers) {
- if (segmentDataManager.getLoadTimeMs() >= reloadJobSubmissionTimestamp) {
+ if (segmentDataManager.getLoadTimeMs() >= reloadJobSubmissionTimestamp
+ || (hasSkippedSegments && jobStatus.isSkipped(segmentDataManager.getSegmentName()))) {
successCount++;
}
}
@@ -96,11 +102,10 @@ public class ControllerJobStatusResource {
.setTotalSegmentCount(totalSegmentCount)
.setSuccessCount(successCount);
- Optional.ofNullable(reloadJobId)
- .map(_serverReloadJobStatusCache::getJobStatus)
- .ifPresent(jobStatus -> response
- .setFailureCount((long) jobStatus.getFailureCount())
- .setSampleSegmentReloadFailures(jobStatus.getFailedSegmentDetails()));
+ if (jobStatus != null) {
+ response.setFailureCount((long) jobStatus.getFailureCount())
+ .setSampleSegmentReloadFailures(jobStatus.getFailedSegmentDetails());
+ }
return JsonUtils.objectToString(response);
} finally {
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java
index 99c23eaa6c6..98fc931230c 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManager.java
@@ -466,12 +466,13 @@ public class HelixInstanceDataManager implements InstanceDataManager {
}
@Override
- public void reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload, String reloadJobId)
+ public void reloadSegment(String tableNameWithType, String segmentName, boolean forceDownload, boolean onlyIfNeeded,
+ String reloadJobId)
throws Exception {
LOGGER.info("Reloading segment: {} in table: {}", segmentName, tableNameWithType);
TableDataManager tableDataManager = _tableDataManagerMap.get(tableNameWithType);
if (tableDataManager != null) {
- tableDataManager.reloadSegment(segmentName, forceDownload, reloadJobId);
+ tableDataManager.reloadSegment(segmentName, forceDownload, onlyIfNeeded, reloadJobId);
} else {
LOGGER.warn("Failed to find data manager for table: {}, skipping reloading segment: {}", tableNameWithType,
segmentName);
@@ -479,12 +480,13 @@ public class HelixInstanceDataManager implements InstanceDataManager {
}
@Override
- public void reloadAllSegments(String tableNameWithType, boolean forceDownload, String reloadJobId)
+ public void reloadAllSegments(String tableNameWithType, boolean forceDownload, boolean onlyIfNeeded,
+ String reloadJobId)
throws Exception {
LOGGER.info("Reloading all segments in table: {}", tableNameWithType);
TableDataManager tableDataManager = _tableDataManagerMap.get(tableNameWithType);
if (tableDataManager != null) {
- tableDataManager.reloadAllSegments(forceDownload, reloadJobId);
+ tableDataManager.reloadAllSegments(forceDownload, onlyIfNeeded, reloadJobId);
} else {
LOGGER.warn("Failed to find data manager for table: {}, skipping reloading all segments", tableNameWithType);
}
@@ -492,12 +494,12 @@ public class HelixInstanceDataManager implements InstanceDataManager {
@Override
public void reloadSegments(String tableNameWithType, List<String> segmentNames, boolean forceDownload,
- String reloadJobId)
+ boolean onlyIfNeeded, String reloadJobId)
throws Exception {
LOGGER.info("Reloading segments: {} in table: {}", segmentNames, tableNameWithType);
TableDataManager tableDataManager = _tableDataManagerMap.get(tableNameWithType);
if (tableDataManager != null) {
- tableDataManager.reloadSegments(segmentNames, forceDownload, reloadJobId);
+ tableDataManager.reloadSegments(segmentNames, forceDownload, onlyIfNeeded, reloadJobId);
} else {
LOGGER.warn("Failed to find data manager for table: {}, skipping reloading segments: {}", tableNameWithType,
segmentNames);
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
index 9312477a3be..ad758a7e868 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/SegmentMessageHandlerFactory.java
@@ -129,6 +129,7 @@ public class SegmentMessageHandlerFactory implements MessageHandlerFactory {
private class SegmentReloadMessageHandler extends DefaultMessageHandler {
private final boolean _forceDownload;
+ private final boolean _onlyIfNeeded;
private final List<String> _segmentList;
private final String _reloadJobId;
@@ -136,6 +137,7 @@ public class SegmentMessageHandlerFactory implements MessageHandlerFactory {
NotificationContext context) {
super(segmentReloadMessage, metrics, context);
_forceDownload = segmentReloadMessage.shouldForceDownload();
+ _onlyIfNeeded = segmentReloadMessage.shouldReloadOnlyIfNeeded();
_segmentList = segmentReloadMessage.getSegmentList();
_reloadJobId = segmentReloadMessage.getReloadJobId();
}
@@ -146,16 +148,18 @@ public class SegmentMessageHandlerFactory implements MessageHandlerFactory {
_logger.info("Handling message: {}", _message);
try {
if (CollectionUtils.isNotEmpty(_segmentList)) {
- _instanceDataManager.reloadSegments(_tableNameWithType, _segmentList, _forceDownload, _reloadJobId);
+ _instanceDataManager.reloadSegments(_tableNameWithType, _segmentList, _forceDownload, _onlyIfNeeded,
+ _reloadJobId);
} else if (StringUtils.isNotEmpty(_segmentName)) {
// TODO: check _segmentName to be backward compatible. Moving forward, we just need to check the list to
// reload one or more segments. If the list or the segment name is empty, all segments are reloaded.
- _instanceDataManager.reloadSegment(_tableNameWithType, _segmentName, _forceDownload, _reloadJobId);
+ _instanceDataManager.reloadSegment(_tableNameWithType, _segmentName, _forceDownload, _onlyIfNeeded,
+ _reloadJobId);
} else {
// NOTE: the method continues if any segment reload encounters an unhandled exception,
// and failed segments are logged out in the end. We don't acquire any permit here as they'll be acquired
// by worked threads later.
- _instanceDataManager.reloadAllSegments(_tableNameWithType, _forceDownload, _reloadJobId);
+ _instanceDataManager.reloadAllSegments(_tableNameWithType, _forceDownload, _onlyIfNeeded, _reloadJobId);
}
helixTaskResult.setSuccess(true);
} catch (Throwable e) {
diff --git a/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ControllerJobStatusResourceTest.java b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ControllerJobStatusResourceTest.java
new file mode 100644
index 00000000000..4d74a07bf9d
--- /dev/null
+++ b/pinot-server/src/test/java/org/apache/pinot/server/api/resources/ControllerJobStatusResourceTest.java
@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.server.api.resources;
+
+import java.util.List;
+import org.apache.pinot.common.response.server.ServerReloadStatusResponse;
+import org.apache.pinot.core.data.manager.InstanceDataManager;
+import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
+import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache;
+import org.apache.pinot.server.starter.ServerInstance;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
+
+public class ControllerJobStatusResourceTest {
+ private static final String TABLE_NAME = "testTable_OFFLINE";
+ private static final long JOB_TIMESTAMP = 1000L;
+
+ private TableDataManager _tableDataManager;
+ private ServerReloadJobStatusCache _jobStatusCache;
+ private ControllerJobStatusResource _resource;
+
+ @BeforeMethod
+ public void setUp() {
+ _tableDataManager = mock(TableDataManager.class);
+ InstanceDataManager instanceDataManager = mock(InstanceDataManager.class);
+ when(instanceDataManager.getTableDataManager(TABLE_NAME)).thenReturn(_tableDataManager);
+ ServerInstance serverInstance = mock(ServerInstance.class);
+ when(serverInstance.getInstanceDataManager()).thenReturn(instanceDataManager);
+ _jobStatusCache = new ServerReloadJobStatusCache("testServer");
+ _resource = new ControllerJobStatusResource(serverInstance, _jobStatusCache);
+ }
+
+ @Test
+ public void testSkippedSegmentsCountedAsSuccessOnce()
+ throws Exception {
+ // reloaded: loaded after the job; skipped: left in place; skippedThenReloaded: skipped but reloaded again later
+ // (counted once); untouched: loaded before the job and not skipped (still pending)
+ List<SegmentDataManager> segments = List.of(segment("reloaded", JOB_TIMESTAMP + 1), segment("skipped", 1L),
+ segment("skippedThenReloaded", JOB_TIMESTAMP), segment("untouched", 1L));
+ when(_tableDataManager.acquireAllSegments()).thenReturn(segments);
+ // A skipped segment that has since been dropped from the server is not counted
+ _jobStatusCache.recordSkipped("job", List.of("skipped", "skippedThenReloaded", "dropped"));
+
+ ServerReloadStatusResponse response = getStatus("job");
+ assertEquals(response.getTotalSegmentCount(), 4);
+ assertEquals(response.getSuccessCount(), 3);
+ assertEquals(response.getFailureCount(), Long.valueOf(0));
+ }
+
+ @Test
+ public void testNoSkippedSegmentsCountsOnlyReloaded()
+ throws Exception {
+ List<SegmentDataManager> segments = List.of(segment("reloaded", JOB_TIMESTAMP + 1), segment("untouched", 1L));
+ when(_tableDataManager.acquireAllSegments()).thenReturn(segments);
+ _jobStatusCache.recordFailure("job", "untouched", new Exception("failed"));
+
+ ServerReloadStatusResponse response = getStatus("job");
+ assertEquals(response.getSuccessCount(), 1);
+ assertEquals(response.getFailureCount(), Long.valueOf(1));
+ }
+
+ private ServerReloadStatusResponse getStatus(String jobId)
+ throws Exception {
+ return JsonUtils.stringToObject(_resource.reloadJobStatus(TABLE_NAME, JOB_TIMESTAMP, null, jobId, null),
+ ServerReloadStatusResponse.class);
+ }
+
+ private static SegmentDataManager segment(String segmentName, long loadTimeMs) {
+ SegmentDataManager segmentDataManager = mock(SegmentDataManager.class);
+ when(segmentDataManager.getSegmentName()).thenReturn(segmentName);
+ when(segmentDataManager.getLoadTimeMs()).thenReturn(loadTimeMs);
+ return segmentDataManager;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org