[GitHub] [incubator-pinot] jtao15 commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…

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

jtao15 commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673285511



##########
File path: pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/merge_rollup/MergeRollupTaskGenerator.java
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.plugin.minion.tasks.merge_rollup;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.helix.task.TaskState;
+import org.apache.pinot.common.lineage.LineageEntry;
+import org.apache.pinot.common.lineage.LineageEntryState;
+import org.apache.pinot.common.lineage.SegmentLineage;
+import org.apache.pinot.common.metadata.segment.OfflineSegmentZKMetadata;
+import org.apache.pinot.common.minion.MergeRollupTaskMetadata;
+import org.apache.pinot.controller.helix.core.minion.ClusterInfoAccessor;
+import org.apache.pinot.controller.helix.core.minion.generator.PinotTaskGenerator;
+import org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.common.MinionConstants.MergeRollupTask;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.spi.annotations.minion.TaskGenerator;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableTaskConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.utils.TimeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A {@link PinotTaskGenerator} implementation for generating tasks of type {@link MergeRollupTask}
+ *
+ * TODO:
+ *  1. Add the support for roll-up
+ *  2. Add the support for customer partitioned table
+ *  3. Add the support for realtime table
+ *
+ * Steps:
+ *
+ *  - Select segments:
+ *    - Fetch all segments, select segments based on segment lineage (removing segmentsFrom for COMPLETED lineage entry and
+ *      segmentsTo for IN_PROGRESS lineage entry)
+ *    - Filter out segments already scheduled by checking zk minion task configs
+ *
+ *  For each granularity (from lowest to highest, e.g. Hourly -> Daily -> Monthly -> Yearly):
+ *    - Calculate merge/rollup window:
+ *      - Read watermarkMs from the {@link MergeRollupTaskMetadata} ZNode
+ *        found at MINION_TASK_METADATA/MergeRollupTaskMetadata/tableNameWithType
+ *        In case of cold-start, no ZNode will exist.
+ *        A new ZNode will be created, with watermarkMs as the smallest time found in all segments truncated to the
+ *        closest bucket start time.
+ *      - The execution window for the task is calculated as,
+ *        windowStartMs = watermarkMs, windowEndMs = windowStartMs + bucketTimeMs
+ *      - Skip scheduling if the window is invalid:
+ *        - If the execution window is not older than bufferTimeMs, no task will be generated
+ *        - The windowEndMs for higher granularity should be less or equal than the waterMarkMs for lower granularity
+ *
+ *    - Select target segments:
+ *      - Filter out segments already merged by checking segment zk metadata {mergeRollupTask.bucketGranularity: granularity}
+ *      - Pick segments overlapping with window [windowStartMs, windowEndMs) up to maxNumRecordsPerTask in sorted order
+ *
+ *    - Bump up waterMarkMs if needed:
+ *      - If there's no segment in the execution window, then everything is merged for the window, bump up the watermark as
+ *        watermarkMs += bucketTimeMs and skip the scheduling.
+ *
+ *    - Create task config
+ */
+@TaskGenerator
+public class MergeRollupTaskGenerator implements PinotTaskGenerator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(MergeRollupTaskGenerator.class);
+
+  private ClusterInfoAccessor _clusterInfoAccessor;
+
+  @Override
+  public void init(ClusterInfoAccessor clusterInfoAccessor) {
+    _clusterInfoAccessor = clusterInfoAccessor;
+  }
+
+  @Override
+  public String getTaskType() {
+    return MergeRollupTask.TASK_TYPE;
+  }
+
+  @Override
+  public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+    String taskType = MergeRollupTask.TASK_TYPE;
+    List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+
+    for (TableConfig tableConfig : tableConfigs) {
+      String offlineTableName = tableConfig.getTableName();
+
+      if (tableConfig.getTableType() != TableType.OFFLINE) {
+        LOGGER.warn("Skip generating task: {} for non-OFFLINE table: {}", taskType, offlineTableName);
+        continue;
+      }
+
+      if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+        LOGGER.warn("Skip generating task: {} for table: {} with customer partition", taskType, offlineTableName);
+        continue;
+      }
+
+      TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
+      Preconditions.checkState(tableTaskConfig != null);
+      Map<String, String> taskConfigs = tableTaskConfig.getConfigsForTaskType(taskType);
+      Preconditions.checkState(taskConfigs != null, "Task config shouldn't be null for table: {}", offlineTableName);
+
+      // Get all segment metadata
+      Set<OfflineSegmentZKMetadata> segmentsForOfflineTable =
+          _clusterInfoAccessor.getOfflineSegmentsMetadata(offlineTableName).stream().collect(Collectors.toSet());
+
+      // Select current segment snapshot based on lineage
+      Set<String> segmentsNotToMerge = new HashSet<>();
+      SegmentLineage segmentLineageForTable = _clusterInfoAccessor.getSegmentLineage(offlineTableName);
+      if (segmentLineageForTable != null) {
+        for (String segmentLineageEntryId : segmentLineageForTable.getLineageEntryIds()) {
+          LineageEntry lineageEntry = segmentLineageForTable.getLineageEntry(segmentLineageEntryId);
+          // Segments shows up on "segmentFrom" field in the lineage entry with "COMPLETED" state cannot be merged.
+          if (lineageEntry.getState() == LineageEntryState.COMPLETED) {
+            segmentsNotToMerge.addAll(lineageEntry.getSegmentsFrom());
+          }
+
+          // Segments shows up on "segmentsTo" field in the lineage entry with "IN_PROGRESS" state cannot be merged.
+          if (lineageEntry.getState() == LineageEntryState.IN_PROGRESS) {
+            segmentsNotToMerge.addAll(lineageEntry.getSegmentsTo());
+          }
+        }
+      }
+
+      LOGGER.info("Start generating task configs for table: {} for task: {}", offlineTableName, taskType);
+
+      Map<String, MergeProperties> mergeProperties = MergeRollupTaskUtils.getAllMergeProperties(taskConfigs);
+      Map<String, PinotTaskConfig> inCompleteGranularities = new HashMap<>();
+
+      // Filter out segments that are already scheduled
+      for (Map.Entry<String, TaskState> entry : TaskGeneratorUtils.getIncompleteTasks(taskType, offlineTableName,
+          _clusterInfoAccessor).entrySet()) {
+        for (PinotTaskConfig taskConfig : _clusterInfoAccessor.getTaskConfigs(entry.getKey())) {
+          inCompleteGranularities.put(taskConfig.getConfigs().get(MergeRollupTask.GRANULARITY_KEY), taskConfig);
+          Arrays.stream(taskConfig.getConfigs()
+              .get(MinionConstants.SEGMENT_NAME_KEY)
+              .split(MinionConstants.SEGMENT_NAME_SEPARATOR)).forEach(s -> segmentsNotToMerge.add(s));
+        }
+      }
+
+      segmentsForOfflineTable = segmentsForOfflineTable.stream()
+          .filter(s -> !segmentsNotToMerge.contains(s.getSegmentName()))
+          .collect(Collectors.toSet());
+      if (segmentsForOfflineTable.isEmpty()) {
+        LOGGER.warn("Skip generating task: {} for table: {}, no segment is found to merge.", taskType,
+            offlineTableName);
+        continue;
+      }
+
+      // From lowest to highest granularity
+      String lowerGranularity = null;
+      long lowerGranularityWatermarkMs = -1;
+      for (Map.Entry<String, MergeProperties> entry : mergeProperties.entrySet()
+          .stream()
+          .sorted((e1, e2) -> Long.compare(TimeUtils.convertPeriodToMillis(e1.getValue().getBucketTimePeriod()),
+              TimeUtils.convertPeriodToMillis(e2.getValue().getBucketTimePeriod())))
+          .collect(Collectors.toList())) {
+        String granularity = entry.getKey();
+        // Only schedule 1 task per granularity

Review comment:
       I'll add a `TODO` here.




-- 
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