navina commented on code in PR #11020:
URL: https://github.com/apache/pinot/pull/11020#discussion_r1254767345
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BaseTableUpsertMetadataManager.java:
##########
@@ -75,6 +108,113 @@ public void init(TableConfig tableConfig, Schema schema,
TableDataManager tableD
_enableSnapshot = upsertConfig.isEnableSnapshot();
_serverMetrics = serverMetrics;
+ _helixManager = helixManager;
+ _segmentPreloadExecutor = segmentPreloadExecutor;
+ try {
+ if (_enableSnapshot && upsertConfig.isEnablePreload()) {
+ // Note that there is an implicit waiting logic between the thread
doing the segment preloading here and the
+ // other helix threads about to process segment state transitions
(e.g. taking segments from OFFLINE to ONLINE).
+ // The thread doing the segment preloading here must complete before
the other helix threads start to handle
+ // segment state transitions. This is ensured implicitly because
segment preloading happens here when
+ // initializing this TableUpsertMetadataManager, which happens when
initializing the TableDataManager, which
+ // happens as the lambda of ConcurrentHashMap.computeIfAbsent()
method, which ensures the waiting logic.
+ _isPreloading = true;
+ preloadSegments();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Got interrupted to preload segments for
table: " + _tableNameWithType, e);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to preload segments for table: " +
_tableNameWithType, e);
+ } finally {
+ _isPreloading = false;
+ }
+ }
+
+ /**
+ * Get the ideal state and find segments assigned to current instance, then
preload those with validDocIds snapshot.
+ * Skip those without the snapshots and those whose crc has changed, as they
will be handled by normal Helix state
+ * transitions, which will proceed after the preloading phase fully
completes.
+ */
+ private void preloadSegments()
+ throws ExecutionException, InterruptedException {
+ LOGGER.info("Preload segments for table: {} fast upsert metadata
recovery", _tableNameWithType);
+ IdealState idealState = HelixHelper.getTableIdealState(_helixManager,
_tableNameWithType);
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
_helixManager.getHelixPropertyStore();
+ String instanceId = getInstanceId();
+ IndexLoadingConfig indexLoadingConfig = createIndexLoadingConfig();
+ List<Future<?>> futures = new ArrayList<>();
+ for (String segmentName : idealState.getPartitionSet()) {
+ Map<String, String> instanceStateMap =
idealState.getInstanceStateMap(segmentName);
+ String state = instanceStateMap.get(instanceId);
+ if
(!CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE.equals(state)) {
+ LOGGER.info("Skip segment: {} as its ideal state: {} is not ONLINE",
segmentName, state);
+ continue;
+ }
+ if (_segmentPreloadExecutor == null) {
+ preloadSegment(segmentName, indexLoadingConfig, propertyStore);
+ continue;
+ }
+ futures.add(_segmentPreloadExecutor.submit(() -> {
+ preloadSegment(segmentName, indexLoadingConfig, propertyStore);
+ }));
+ }
+ for (Future<?> f : futures) {
+ f.get();
Review Comment:
yes. As you explained, we can gracefully fallback to normal segment loading
, if the preloading of a segment fails for whatever reason.
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
