[GitHub] [pinot] sajjad-moradi commented on a diff in pull request #8786: Misc fixes on segment validation for uploaded real-time segments

[GitBox <gi...@apache.org>]

sajjad-moradi commented on code in PR #8786:
URL: https://github.com/apache/pinot/pull/8786#discussion_r883966738


##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -59,6 +59,28 @@ public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber
     _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
   }
 
+  private LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, String creationTime,
+      String segmentName) {
+    _tableName = tableName;
+    _partitionGroupId = partitionGroupId;
+    _sequenceNumber = sequenceNumber;
+    _creationTime = creationTime;
+    _segmentName = segmentName;
+  }
+
+  /**
+   * Returns the {@link LLCSegmentName} for the given segment name, or {@code null} if the given segment name does not
+   * represent an LLC segment.
+   */
+  @Nullable
+  public static LLCSegmentName getLLCSegmentName(String segmentName) {
+    String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
+    if (parts.length != 4) {
+      return null;
+    }
+    return new LLCSegmentName(parts[0], Integer.parseInt(parts[1]), Integer.parseInt(parts[2]), parts[3], segmentName);

Review Comment:
   Why not just `return new LLCSegmentName(segmentName)`?
   This way there's no need to create the new private constructor.
   Also could you rename the static method? `LLCSegmentName.getLCCSegmentName()` is a bit redundant. How about `LLCSegmentName.of()`?



##########
pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java:
##########
@@ -90,86 +88,83 @@ protected Context preprocess(Properties periodicTaskProperties) {
 
   @Override
   protected void processTable(String tableNameWithType, Context context) {
-    TableType tableType = TableNameBuilder.getTableTypeFromTableName(tableNameWithType);
-    if (tableType == TableType.REALTIME) {
+    if (!TableNameBuilder.isRealtimeTableResource(tableNameWithType)) {
+      return;
+    }
 
-      TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
-      if (tableConfig == null) {
-        LOGGER.warn("Failed to find table config for table: {}, skipping validation", tableNameWithType);
-        return;
-      }
+    TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+    if (tableConfig == null) {
+      LOGGER.warn("Failed to find table config for table: {}, skipping validation", tableNameWithType);
+      return;
+    }
+    PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
+        IngestionConfigUtils.getStreamConfigMap(tableConfig));
 
-      if (context._runSegmentLevelValidation) {
-        runSegmentLevelValidation(tableConfig);
-      }
+    if (context._runSegmentLevelValidation) {
+      runSegmentLevelValidation(tableConfig, streamConfig);
+    }
 
-      PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
-          IngestionConfigUtils.getStreamConfigMap(tableConfig));
-      if (streamConfig.hasLowLevelConsumerType()) {
-        _llcRealtimeSegmentManager.ensureAllPartitionsConsuming(tableConfig,
-            streamConfig, context._recreateDeletedConsumingSegment);
-      }
+    if (streamConfig.hasLowLevelConsumerType()) {
+      _llcRealtimeSegmentManager.ensureAllPartitionsConsuming(tableConfig, streamConfig,
+          context._recreateDeletedConsumingSegment);
     }
   }
 
-  private void runSegmentLevelValidation(TableConfig tableConfig) {
+  private void runSegmentLevelValidation(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig) {
     String realtimeTableName = tableConfig.getTableName();
     List<SegmentZKMetadata> segmentsZKMetadata = _pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName);
-    boolean countHLCSegments = true;  // false if this table has ONLY LLC segments (i.e. fully migrated)
-    StreamConfig streamConfig =
-        new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(tableConfig));
-    if (streamConfig.hasLowLevelConsumerType() && !streamConfig.hasHighLevelConsumerType()) {
-      countHLCSegments = false;
-    }
-    // Update the gauge to contain the total document count in the segments
-    _validationMetrics.updateTotalDocumentCountGauge(tableConfig.getTableName(),
-        computeRealtimeTotalDocumentInSegments(segmentsZKMetadata, countHLCSegments));
 
-    if (streamConfig.hasLowLevelConsumerType()
-        && _llcRealtimeSegmentManager.isDeepStoreLLCSegmentUploadRetryEnabled()) {
+    // Update the total document count gauge
+    // Count HLC segments if there is no LLC consumer configured
+    boolean hasLLC = streamConfig.hasLowLevelConsumerType();
+    _validationMetrics.updateTotalDocumentCountGauge(realtimeTableName,
+        computeTotalDocumentCount(segmentsZKMetadata, hasLLC));

Review Comment:
   So the assumption here is that there's no table table with both HLC's and LLC's, right? If there's such scenario, this changes is backward incompatible.



##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/readerwriter/RealtimeIndexOffHeapMemoryManager.java:
##########
@@ -51,8 +51,8 @@ public abstract class RealtimeIndexOffHeapMemoryManager implements PinotDataBuff
   protected RealtimeIndexOffHeapMemoryManager(ServerMetrics serverMetrics, String segmentName) {
     _serverMetrics = serverMetrics;
     _segmentName = segmentName;
-    if (SegmentName.isLowLevelConsumerSegmentName(segmentName)) {
-      LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
+    LLCSegmentName llcSegmentName = LLCSegmentName.getLLCSegmentName(segmentName);
+    if (llcSegmentName != null) {
       _tableName = llcSegmentName.getTableName();

Review Comment:
   ```suggestion
       if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
         _tableName = LLCSegmentName.of(segmentName).getTableName();
   ```



##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGenerator.java:
##########
@@ -243,41 +243,42 @@ public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
   }
 
   /**
-   * Fetch completed (non-consuming) segment and partition information
+   * Fetch completed (DONE/UPLOADED) segment and partition information
+   *
    * @param realtimeTableName the realtime table name
-   * @param completedSegmentsZKMetadata list for collecting the completed segments ZK metadata
-   * @param partitionToLatestCompletedSegmentName map for collecting the partitionId to the latest completed segment
-   *                                              name
+   * @param completedSegmentsZKMetadata list for collecting the completed (DONE/UPLOADED) segments ZK metadata
+   * @param partitionToLatestLLCSegmentName map for collecting the partitionId to the latest LLC segment name
    * @param allPartitions set for collecting all partition ids
    */
   private void getCompletedSegmentsInfo(String realtimeTableName, List<SegmentZKMetadata> completedSegmentsZKMetadata,
-      Map<Integer, String> partitionToLatestCompletedSegmentName, Set<Integer> allPartitions) {
+      Map<Integer, String> partitionToLatestLLCSegmentName, Set<Integer> allPartitions) {
     List<SegmentZKMetadata> segmentsZKMetadata = _clusterInfoAccessor.getSegmentsZKMetadata(realtimeTableName);
 
     Map<Integer, LLCSegmentName> latestLLCSegmentNameMap = new HashMap<>();
     for (SegmentZKMetadata segmentZKMetadata : segmentsZKMetadata) {
-      LLCSegmentName llcSegmentName = new LLCSegmentName(segmentZKMetadata.getSegmentName());
-      allPartitions.add(llcSegmentName.getPartitionGroupId());
-
-      if (segmentZKMetadata.getStatus().equals(Segment.Realtime.Status.DONE)) {
+      Segment.Realtime.Status status = segmentZKMetadata.getStatus();
+      if (status.isCompleted()) {
         completedSegmentsZKMetadata.add(segmentZKMetadata);
-        latestLLCSegmentNameMap.compute(llcSegmentName.getPartitionGroupId(),
-            (partitionGroupId, latestLLCSegmentName) -> {
-              if (latestLLCSegmentName == null) {
-                return llcSegmentName;
-              } else {
-                if (llcSegmentName.getSequenceNumber() > latestLLCSegmentName.getSequenceNumber()) {
-                  return llcSegmentName;
-                } else {
-                  return latestLLCSegmentName;
-                }
-              }
-            });
+      }
+      LLCSegmentName llcSegmentName = LLCSegmentName.getLLCSegmentName(segmentZKMetadata.getSegmentName());
+      if (llcSegmentName != null) {

Review Comment:
   Can you add a comment after if statement like: `// UPLOADED segments that don't conform to llc segment names`



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org