You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2021/07/19 16:59:39 UTC
[GitHub] [incubator-pinot] jtao15 opened a new pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
jtao15 opened a new pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178
Merge/Rollup task scheduler which supports multi-level CONCAT tasks for offline tables.
1. Add MergeRollupTaskGenerator and unit test.
2. Change minion constants and util functions based on updated [design](https://docs.google.com/document/d/1-AKCfXNXdoNjFIvJ87wjWwFM_38gS0NCwFrIYjYsqp8/edit#)
Plan to add integration test once time partition and segment name prefix are supported for MergeRollupTaskExecutor
--
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
[GitHub] [incubator-pinot] jtao15 commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
jtao15 commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673356409
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
We assume all segments before the watermark are merged, so it won't have a chance to be merged. Hmm, why would the user want to upload some old data? If they want to backfill 07/15 data, they should replace all segments on 07/15. For now, we are not covering the case if they backfill a large number of segments on 07/15.
--
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
[GitHub] [incubator-pinot] jtao15 commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
jtao15 commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882707232
@snleee
--
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
[GitHub] [incubator-pinot] snleee commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882730722
Please fix the test issue. It looks that it has a compilation issue in `MergeRollupTaskGeneratorTest .java`
--
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
[GitHub] [incubator-pinot] jtao15 commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
jtao15 commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673285942
##########
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
+ if (inCompleteGranularities.containsKey(granularity)) {
+ LOGGER.info("Found incomplete tasks: {} for same granularity: {} and table: {}. Skipping task generation.",
+ inCompleteGranularities.get(granularity), granularity, offlineTableName);
+ continue;
+ }
+
+ MergeProperties mergeProperty = entry.getValue();
+ Set<OfflineSegmentZKMetadata> candidateSegments = new HashSet<>(segmentsForOfflineTable);
+
+ // Get the bucket size and buffer
+ long bucketMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBucketTimePeriod());
+ long bufferMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBufferTimePeriod());
+
+ // Get watermark from MergeRollupTaskMetadata ZNode. WindowStart = watermark. WindowEnd = windowStart + bucket.
+ long windowStartMs = getWatermarkMs(offlineTableName, candidateSegments, bucketMs, granularity);
+ long windowEndMs = windowStartMs + bucketMs;
+
+ // Check that execution window endTimeMs <= now - bufferTime
+ if (windowEndMs > System.currentTimeMillis() - bufferMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} is not older than buffer time: {} configured as {} ago. Skipping task generation: {}",
+ windowStartMs, windowEndMs, bufferMs, bufferMs, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Check that execution window endTimeMs <= waterMark of the lower granularity
+ if (lowerGranularity != null && windowEndMs > lowerGranularityWatermarkMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} of granularity: {} is not older than the watermark: {} of granularity: (). Skipping task generation: {}",
+ windowStartMs, windowEndMs, granularity, lowerGranularityWatermarkMs, lowerGranularity, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Filter out segments that are already merged for the granularity
Review comment:
The segment zk metadata is updated when uploading the merged segment.
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673317663
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
I think that it makes sense that we need to know about the current window's start & end time in order to mark the segments.
What happens if some segments that are older than the watermark get pushed? (e.g. watermark=`0720`, segment=`0715_0715`) I think that we haven't covered this case on how we handle it. @Jackie-Jiang @jtao15
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673317663
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
I think that it makes sense that we need to know about the current window's start & end time in order to mark the segments.
What happens if some segments that are older than the watermark get pushed? (e.g. watermark=`0720`, segment=`0715_0715`) I think that we haven't covered this case. @Jackie-Jiang @jtao15
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673329268
##########
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);
Review comment:
Can you double-check if this covers `REFRESH` use cases?
--
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
[GitHub] [incubator-pinot] codecov-commenter edited a comment on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882751873
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#7178](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (4611426) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/dbaac00b6f77161f079f07ff7d337cbde55d2cfd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (dbaac00) will **decrease** coverage by `8.25%`.
> The diff coverage is `89.39%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #7178 +/- ##
============================================
- Coverage 73.57% 65.31% -8.26%
Complexity 92 92
============================================
Files 1500 1501 +1
Lines 73758 73928 +170
Branches 10637 10670 +33
============================================
- Hits 54264 48285 -5979
- Misses 15955 22224 +6269
+ Partials 3539 3419 -120
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `?` | |
| unittests | `65.31% <89.39%> (+<0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...troller/helix/core/minion/ClusterInfoAccessor.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL21pbmlvbi9DbHVzdGVySW5mb0FjY2Vzc29yLmphdmE=) | `30.00% <0.00%> (-41.43%)` | :arrow_down: |
| [.../org/apache/pinot/core/common/MinionConstants.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9jb21tb24vTWluaW9uQ29uc3RhbnRzLmphdmE=) | `0.00% <ø> (ø)` | |
| [...e/pinot/common/minion/MergeRollupTaskMetadata.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWluaW9uL01lcmdlUm9sbHVwVGFza01ldGFkYXRhLmphdmE=) | `23.80% <20.00%> (+23.80%)` | :arrow_up: |
| [...inion/tasks/merge\_rollup/MergeRollupTaskUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza1V0aWxzLmphdmE=) | `83.01% <76.92%> (-4.49%)` | :arrow_down: |
| [...n/tasks/merge\_rollup/MergeRollupTaskGenerator.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza0dlbmVyYXRvci5qYXZh) | `94.30% <94.30%> (ø)` | |
| [...gin/minion/tasks/merge\_rollup/MergeProperties.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUHJvcGVydGllcy5qYXZh) | `100.00% <100.00%> (ø)` | |
| [...a/org/apache/pinot/common/metrics/MinionMeter.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9NaW5pb25NZXRlci5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [.../apache/pinot/common/metrics/BrokerQueryPhase.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9Ccm9rZXJRdWVyeVBoYXNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [.../apache/pinot/common/metrics/MinionQueryPhase.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9NaW5pb25RdWVyeVBoYXNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...pache/pinot/common/utils/grpc/GrpcQueryClient.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdXRpbHMvZ3JwYy9HcnBjUXVlcnlDbGllbnQuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [361 more](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [dbaac00...4611426](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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
[GitHub] [incubator-pinot] codecov-commenter commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882751873
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#7178](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b9881eb) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/dbaac00b6f77161f079f07ff7d337cbde55d2cfd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (dbaac00) will **decrease** coverage by `31.80%`.
> The diff coverage is `2.52%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #7178 +/- ##
=============================================
- Coverage 73.57% 41.76% -31.81%
+ Complexity 92 7 -85
=============================================
Files 1500 1500
Lines 73758 73921 +163
Branches 10637 10668 +31
=============================================
- Hits 54264 30873 -23391
- Misses 15955 40425 +24470
+ Partials 3539 2623 -916
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `41.76% <2.52%> (-0.10%)` | :arrow_down: |
| unittests | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...e/pinot/common/minion/MergeRollupTaskMetadata.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWluaW9uL01lcmdlUm9sbHVwVGFza01ldGFkYXRhLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...troller/helix/core/minion/ClusterInfoAccessor.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL21pbmlvbi9DbHVzdGVySW5mb0FjY2Vzc29yLmphdmE=) | `53.33% <0.00%> (-18.10%)` | :arrow_down: |
| [.../org/apache/pinot/core/common/MinionConstants.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9jb21tb24vTWluaW9uQ29uc3RhbnRzLmphdmE=) | `0.00% <ø> (ø)` | |
| [...gin/minion/tasks/merge\_rollup/MergeProperties.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUHJvcGVydGllcy5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...inion/tasks/merge\_rollup/MergeRollupTaskUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza1V0aWxzLmphdmE=) | `0.00% <0.00%> (-87.50%)` | :arrow_down: |
| [...n/tasks/merge\_rollup/MergeRollupTaskGenerator.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza0dlbmVyYXRvci5qYXZh) | `3.16% <3.16%> (ø)` | |
| [...c/main/java/org/apache/pinot/common/tier/Tier.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdGllci9UaWVyLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ava/org/apache/pinot/spi/data/MetricFieldSpec.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvZGF0YS9NZXRyaWNGaWVsZFNwZWMuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...va/org/apache/pinot/spi/utils/BigDecimalUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvdXRpbHMvQmlnRGVjaW1hbFV0aWxzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...java/org/apache/pinot/common/tier/TierFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdGllci9UaWVyRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [953 more](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [dbaac00...b9881eb](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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
[GitHub] [incubator-pinot] jtao15 commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
jtao15 commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673296717
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
We will not filter the records, while the start/end time are useful to mark the segment as merged by setting customMap as `{mergeRollupTask.bucketGranularity : granularityLevel}`. Say we have segments: `s1_0719_0720`, `s2_0720_0720` and we use 1 day bucket time. The first run merge `s1_0719_0720` to `merged_s3_0719_0719` and `merged_s4_0720_0720`. `merged_s3_0719_0719` will be marked since it's in the target window and won't be picked for the next merge task. `merged_s4_0720_0720` will not be marked. The second merge task will pick `merged_s4_0720_0720` and `s2_0720_0720`.
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673317663
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
I think that it makes sense that we need to know about the current window's start & end time in order to mark the segments.
What happens if some segments that are older than the watermark get pushed? (e.g. watermark=`0720`, newly pushed segments=`0715_0715`) I think that we haven't covered this case. @Jackie-Jiang @jtao15
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r672645646
##########
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()
Review comment:
We use `SegmentLineageBasedSegmentPreSelector.preSelect()` instead of explicitly keeping track of `segmentsNotToMerge`
##########
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()
Review comment:
Computing the entryList is too long and it's hard to understand the logic. Can you compute the entryList first in a separate line and add a bit more comment on what you're trying to compute here?
##########
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);
Review comment:
We should move this line up and put after L124 (after checking skipping conditions)
##########
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
+ if (inCompleteGranularities.containsKey(granularity)) {
+ LOGGER.info("Found incomplete tasks: {} for same granularity: {} and table: {}. Skipping task generation.",
+ inCompleteGranularities.get(granularity), granularity, offlineTableName);
+ continue;
+ }
+
+ MergeProperties mergeProperty = entry.getValue();
+ Set<OfflineSegmentZKMetadata> candidateSegments = new HashSet<>(segmentsForOfflineTable);
+
+ // Get the bucket size and buffer
+ long bucketMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBucketTimePeriod());
+ long bufferMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBufferTimePeriod());
+
+ // Get watermark from MergeRollupTaskMetadata ZNode. WindowStart = watermark. WindowEnd = windowStart + bucket.
+ long windowStartMs = getWatermarkMs(offlineTableName, candidateSegments, bucketMs, granularity);
+ long windowEndMs = windowStartMs + bucketMs;
+
+ // Check that execution window endTimeMs <= now - bufferTime
+ if (windowEndMs > System.currentTimeMillis() - bufferMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} is not older than buffer time: {} configured as {} ago. Skipping task generation: {}",
+ windowStartMs, windowEndMs, bufferMs, bufferMs, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Check that execution window endTimeMs <= waterMark of the lower granularity
+ if (lowerGranularity != null && windowEndMs > lowerGranularityWatermarkMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} of granularity: {} is not older than the watermark: {} of granularity: (). Skipping task generation: {}",
+ windowStartMs, windowEndMs, granularity, lowerGranularityWatermarkMs, lowerGranularity, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Filter out segments that are already merged for the granularity
+ candidateSegments = candidateSegments.stream()
+ .filter(s -> s.getCustomMap() != null ? !granularity.equalsIgnoreCase(
+ s.getCustomMap().get(taskType + MinionConstants.TASK_BUCKET_GRANULARITY_SUFFIX)) : true)
+ .collect(Collectors.toSet());
+
+ // Find segments with data overlapping execution window: windowStart (inclusive) to windowEnd (exclusive),
+ // sort segments based on start time, if start time is the same, sort on end time. Pick segments up to
+ // maxNumRecordsPerTask (soft boundary).
+ List<OfflineSegmentZKMetadata> targetSegments = new ArrayList<>();
+ for (OfflineSegmentZKMetadata offlineSegmentZKMetadata : candidateSegments) {
+ // check overlap with the window
+ if (windowStartMs <= offlineSegmentZKMetadata.getEndTimeMs()
+ && offlineSegmentZKMetadata.getStartTimeMs() < windowEndMs) {
+ targetSegments.add(offlineSegmentZKMetadata);
+ }
+ }
+
+ if (targetSegments.isEmpty()) {
+ LOGGER.info("No segments overlapping with window [{}, {}), Skipping task generation: {}", windowStartMs,
+ windowEndMs, taskType);
+ // Bump up watermark
+ updateWatermarkMs(offlineTableName, granularity, windowStartMs + bucketMs);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs + bucketMs;
+ continue;
+ }
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+
+ Collections.sort(targetSegments, (a, b) -> {
Review comment:
Can you add the comment on what are we sorting based on?
##########
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<>();
Review comment:
The same segment filtering logic is written in `SegmentLineageBasedSegmentPreSelector`
Can we share the filtering logic? We will need to modify slightly the pre-selector as well.
##########
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
+ if (inCompleteGranularities.containsKey(granularity)) {
+ LOGGER.info("Found incomplete tasks: {} for same granularity: {} and table: {}. Skipping task generation.",
+ inCompleteGranularities.get(granularity), granularity, offlineTableName);
+ continue;
+ }
+
+ MergeProperties mergeProperty = entry.getValue();
+ Set<OfflineSegmentZKMetadata> candidateSegments = new HashSet<>(segmentsForOfflineTable);
+
+ // Get the bucket size and buffer
+ long bucketMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBucketTimePeriod());
+ long bufferMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBufferTimePeriod());
+
+ // Get watermark from MergeRollupTaskMetadata ZNode. WindowStart = watermark. WindowEnd = windowStart + bucket.
+ long windowStartMs = getWatermarkMs(offlineTableName, candidateSegments, bucketMs, granularity);
+ long windowEndMs = windowStartMs + bucketMs;
+
+ // Check that execution window endTimeMs <= now - bufferTime
+ if (windowEndMs > System.currentTimeMillis() - bufferMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} is not older than buffer time: {} configured as {} ago. Skipping task generation: {}",
+ windowStartMs, windowEndMs, bufferMs, bufferMs, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Check that execution window endTimeMs <= waterMark of the lower granularity
+ if (lowerGranularity != null && windowEndMs > lowerGranularityWatermarkMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} of granularity: {} is not older than the watermark: {} of granularity: (). Skipping task generation: {}",
+ windowStartMs, windowEndMs, granularity, lowerGranularityWatermarkMs, lowerGranularity, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Filter out segments that are already merged for the granularity
Review comment:
Where do we update the segment zk metadata?
##########
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:
Are we planning to support subtasks (i.e. multiple tasks for a single time bucket using `MAX_NUM_RECORDS_PER_TASK`? If we plan to add it in the next stage, please indicate that as `TODO` comment.
Ideally, we should compute 1 time bucket per granularity and multiple tasks within a bucket is allowed (e.g. splitting a work into multiple tasks for 1 month time bucket)
--
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
[GitHub] [incubator-pinot] jtao15 commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
jtao15 commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673286059
##########
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<>();
Review comment:
Yeah, we should be able to share the logic. Let me see how can I reuse the pre-selector.
--
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
[GitHub] [incubator-pinot] codecov-commenter commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882751873
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#7178](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b9881eb) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/dbaac00b6f77161f079f07ff7d337cbde55d2cfd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (dbaac00) will **decrease** coverage by `31.80%`.
> The diff coverage is `2.52%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #7178 +/- ##
=============================================
- Coverage 73.57% 41.76% -31.81%
+ Complexity 92 7 -85
=============================================
Files 1500 1500
Lines 73758 73921 +163
Branches 10637 10668 +31
=============================================
- Hits 54264 30873 -23391
- Misses 15955 40425 +24470
+ Partials 3539 2623 -916
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `41.76% <2.52%> (-0.10%)` | :arrow_down: |
| unittests | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...e/pinot/common/minion/MergeRollupTaskMetadata.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWluaW9uL01lcmdlUm9sbHVwVGFza01ldGFkYXRhLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...troller/helix/core/minion/ClusterInfoAccessor.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL21pbmlvbi9DbHVzdGVySW5mb0FjY2Vzc29yLmphdmE=) | `53.33% <0.00%> (-18.10%)` | :arrow_down: |
| [.../org/apache/pinot/core/common/MinionConstants.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9jb21tb24vTWluaW9uQ29uc3RhbnRzLmphdmE=) | `0.00% <ø> (ø)` | |
| [...gin/minion/tasks/merge\_rollup/MergeProperties.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUHJvcGVydGllcy5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...inion/tasks/merge\_rollup/MergeRollupTaskUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza1V0aWxzLmphdmE=) | `0.00% <0.00%> (-87.50%)` | :arrow_down: |
| [...n/tasks/merge\_rollup/MergeRollupTaskGenerator.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza0dlbmVyYXRvci5qYXZh) | `3.16% <3.16%> (ø)` | |
| [...c/main/java/org/apache/pinot/common/tier/Tier.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdGllci9UaWVyLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ava/org/apache/pinot/spi/data/MetricFieldSpec.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvZGF0YS9NZXRyaWNGaWVsZFNwZWMuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...va/org/apache/pinot/spi/utils/BigDecimalUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvdXRpbHMvQmlnRGVjaW1hbFV0aWxzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...java/org/apache/pinot/common/tier/TierFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdGllci9UaWVyRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [953 more](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [dbaac00...b9881eb](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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
[GitHub] [incubator-pinot] codecov-commenter edited a comment on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882751873
--
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
[GitHub] [incubator-pinot] jtao15 commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
jtao15 commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882707232
@snleee
--
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
[GitHub] [incubator-pinot] snleee commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882730722
Please fix the test issue. It looks that it has a compilation issue in `MergeRollupTaskGeneratorTest .java`
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r672645646
##########
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()
Review comment:
We use `SegmentLineageBasedSegmentPreSelector.preSelect()` instead of explicitly keeping track of `segmentsNotToMerge`
##########
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()
Review comment:
Computing the entryList is too long and it's hard to understand the logic. Can you compute the entryList first in a separate line and add a bit more comment on what you're trying to compute here?
##########
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);
Review comment:
We should move this line up and put after L124 (after checking skipping conditions)
##########
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
+ if (inCompleteGranularities.containsKey(granularity)) {
+ LOGGER.info("Found incomplete tasks: {} for same granularity: {} and table: {}. Skipping task generation.",
+ inCompleteGranularities.get(granularity), granularity, offlineTableName);
+ continue;
+ }
+
+ MergeProperties mergeProperty = entry.getValue();
+ Set<OfflineSegmentZKMetadata> candidateSegments = new HashSet<>(segmentsForOfflineTable);
+
+ // Get the bucket size and buffer
+ long bucketMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBucketTimePeriod());
+ long bufferMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBufferTimePeriod());
+
+ // Get watermark from MergeRollupTaskMetadata ZNode. WindowStart = watermark. WindowEnd = windowStart + bucket.
+ long windowStartMs = getWatermarkMs(offlineTableName, candidateSegments, bucketMs, granularity);
+ long windowEndMs = windowStartMs + bucketMs;
+
+ // Check that execution window endTimeMs <= now - bufferTime
+ if (windowEndMs > System.currentTimeMillis() - bufferMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} is not older than buffer time: {} configured as {} ago. Skipping task generation: {}",
+ windowStartMs, windowEndMs, bufferMs, bufferMs, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Check that execution window endTimeMs <= waterMark of the lower granularity
+ if (lowerGranularity != null && windowEndMs > lowerGranularityWatermarkMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} of granularity: {} is not older than the watermark: {} of granularity: (). Skipping task generation: {}",
+ windowStartMs, windowEndMs, granularity, lowerGranularityWatermarkMs, lowerGranularity, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Filter out segments that are already merged for the granularity
+ candidateSegments = candidateSegments.stream()
+ .filter(s -> s.getCustomMap() != null ? !granularity.equalsIgnoreCase(
+ s.getCustomMap().get(taskType + MinionConstants.TASK_BUCKET_GRANULARITY_SUFFIX)) : true)
+ .collect(Collectors.toSet());
+
+ // Find segments with data overlapping execution window: windowStart (inclusive) to windowEnd (exclusive),
+ // sort segments based on start time, if start time is the same, sort on end time. Pick segments up to
+ // maxNumRecordsPerTask (soft boundary).
+ List<OfflineSegmentZKMetadata> targetSegments = new ArrayList<>();
+ for (OfflineSegmentZKMetadata offlineSegmentZKMetadata : candidateSegments) {
+ // check overlap with the window
+ if (windowStartMs <= offlineSegmentZKMetadata.getEndTimeMs()
+ && offlineSegmentZKMetadata.getStartTimeMs() < windowEndMs) {
+ targetSegments.add(offlineSegmentZKMetadata);
+ }
+ }
+
+ if (targetSegments.isEmpty()) {
+ LOGGER.info("No segments overlapping with window [{}, {}), Skipping task generation: {}", windowStartMs,
+ windowEndMs, taskType);
+ // Bump up watermark
+ updateWatermarkMs(offlineTableName, granularity, windowStartMs + bucketMs);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs + bucketMs;
+ continue;
+ }
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+
+ Collections.sort(targetSegments, (a, b) -> {
Review comment:
Can you add the comment on what are we sorting based on?
##########
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<>();
Review comment:
The same segment filtering logic is written in `SegmentLineageBasedSegmentPreSelector`
Can we share the filtering logic? We will need to modify slightly the pre-selector as well.
##########
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
+ if (inCompleteGranularities.containsKey(granularity)) {
+ LOGGER.info("Found incomplete tasks: {} for same granularity: {} and table: {}. Skipping task generation.",
+ inCompleteGranularities.get(granularity), granularity, offlineTableName);
+ continue;
+ }
+
+ MergeProperties mergeProperty = entry.getValue();
+ Set<OfflineSegmentZKMetadata> candidateSegments = new HashSet<>(segmentsForOfflineTable);
+
+ // Get the bucket size and buffer
+ long bucketMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBucketTimePeriod());
+ long bufferMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBufferTimePeriod());
+
+ // Get watermark from MergeRollupTaskMetadata ZNode. WindowStart = watermark. WindowEnd = windowStart + bucket.
+ long windowStartMs = getWatermarkMs(offlineTableName, candidateSegments, bucketMs, granularity);
+ long windowEndMs = windowStartMs + bucketMs;
+
+ // Check that execution window endTimeMs <= now - bufferTime
+ if (windowEndMs > System.currentTimeMillis() - bufferMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} is not older than buffer time: {} configured as {} ago. Skipping task generation: {}",
+ windowStartMs, windowEndMs, bufferMs, bufferMs, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Check that execution window endTimeMs <= waterMark of the lower granularity
+ if (lowerGranularity != null && windowEndMs > lowerGranularityWatermarkMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} of granularity: {} is not older than the watermark: {} of granularity: (). Skipping task generation: {}",
+ windowStartMs, windowEndMs, granularity, lowerGranularityWatermarkMs, lowerGranularity, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Filter out segments that are already merged for the granularity
Review comment:
Where do we update the segment zk metadata?
##########
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:
Are we planning to support subtasks (i.e. multiple tasks for a single time bucket using `MAX_NUM_RECORDS_PER_TASK`? If we plan to add it in the next stage, please indicate that as `TODO` comment.
Ideally, we should compute 1 time bucket per granularity and multiple tasks within a bucket is allowed (e.g. splitting a work into multiple tasks for 1 month time bucket)
--
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
[GitHub] [incubator-pinot] codecov-commenter commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882751873
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#7178](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b9881eb) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/dbaac00b6f77161f079f07ff7d337cbde55d2cfd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (dbaac00) will **decrease** coverage by `31.80%`.
> The diff coverage is `2.52%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #7178 +/- ##
=============================================
- Coverage 73.57% 41.76% -31.81%
+ Complexity 92 7 -85
=============================================
Files 1500 1500
Lines 73758 73921 +163
Branches 10637 10668 +31
=============================================
- Hits 54264 30873 -23391
- Misses 15955 40425 +24470
+ Partials 3539 2623 -916
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `41.76% <2.52%> (-0.10%)` | :arrow_down: |
| unittests | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...e/pinot/common/minion/MergeRollupTaskMetadata.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWluaW9uL01lcmdlUm9sbHVwVGFza01ldGFkYXRhLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...troller/helix/core/minion/ClusterInfoAccessor.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL21pbmlvbi9DbHVzdGVySW5mb0FjY2Vzc29yLmphdmE=) | `53.33% <0.00%> (-18.10%)` | :arrow_down: |
| [.../org/apache/pinot/core/common/MinionConstants.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9jb21tb24vTWluaW9uQ29uc3RhbnRzLmphdmE=) | `0.00% <ø> (ø)` | |
| [...gin/minion/tasks/merge\_rollup/MergeProperties.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUHJvcGVydGllcy5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...inion/tasks/merge\_rollup/MergeRollupTaskUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza1V0aWxzLmphdmE=) | `0.00% <0.00%> (-87.50%)` | :arrow_down: |
| [...n/tasks/merge\_rollup/MergeRollupTaskGenerator.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza0dlbmVyYXRvci5qYXZh) | `3.16% <3.16%> (ø)` | |
| [...c/main/java/org/apache/pinot/common/tier/Tier.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdGllci9UaWVyLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ava/org/apache/pinot/spi/data/MetricFieldSpec.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvZGF0YS9NZXRyaWNGaWVsZFNwZWMuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...va/org/apache/pinot/spi/utils/BigDecimalUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvdXRpbHMvQmlnRGVjaW1hbFV0aWxzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...java/org/apache/pinot/common/tier/TierFactory.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdGllci9UaWVyRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [953 more](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [dbaac00...b9881eb](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673582170
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
There's a valid use case:
1. user pushes data for 07/15
2. pinot merges segments for 07/15
3. user finds out the issue and backfill 07/15 later
4. it's good if pinot merges backfilled segments for 07/15 but we won't do it with the current design since the watermark is already past those segments so we don't pick them. As we discussed today, one solution is to open up the way for triggering ad-hoc merge-rollup job for a specific window.
--
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
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r672670245
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
For merge/rollup task, we should not use the window start/end, but use the partition bucket instead. Different from the realtime-to-offline task, we need to keep all the records (cannot filter the records) for merge/rollup task.
--
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
[GitHub] [incubator-pinot] jtao15 commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
jtao15 commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882707232
@snleee
--
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
[GitHub] [incubator-pinot] codecov-commenter edited a comment on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882751873
--
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
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r672670245
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
For merge/rollup task, we should not use the window start/end, but use the partition bucket instead. Different from the realtime-to-offline task, we need to keep all the records (cannot filter the records) for merge/rollup task.
--
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
[GitHub] [incubator-pinot] codecov-commenter edited a comment on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882751873
# [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#7178](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (4611426) into [master](https://codecov.io/gh/apache/incubator-pinot/commit/dbaac00b6f77161f079f07ff7d337cbde55d2cfd?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (dbaac00) will **decrease** coverage by `0.00%`.
> The diff coverage is `89.39%`.
[](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #7178 +/- ##
============================================
- Coverage 73.57% 73.56% -0.01%
Complexity 92 92
============================================
Files 1500 1501 +1
Lines 73758 73928 +170
Branches 10637 10670 +33
============================================
+ Hits 54264 54385 +121
- Misses 15955 15998 +43
- Partials 3539 3545 +6
```
| Flag | Coverage Δ | |
|---|---|---|
| integration | `41.90% <2.52%> (+0.05%)` | :arrow_up: |
| unittests | `65.31% <89.39%> (+<0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...troller/helix/core/minion/ClusterInfoAccessor.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL21pbmlvbi9DbHVzdGVySW5mb0FjY2Vzc29yLmphdmE=) | `66.66% <0.00%> (-4.77%)` | :arrow_down: |
| [.../org/apache/pinot/core/common/MinionConstants.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9jb21tb24vTWluaW9uQ29uc3RhbnRzLmphdmE=) | `0.00% <ø> (ø)` | |
| [...e/pinot/common/minion/MergeRollupTaskMetadata.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWluaW9uL01lcmdlUm9sbHVwVGFza01ldGFkYXRhLmphdmE=) | `23.80% <20.00%> (+23.80%)` | :arrow_up: |
| [...inion/tasks/merge\_rollup/MergeRollupTaskUtils.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza1V0aWxzLmphdmE=) | `83.01% <76.92%> (-4.49%)` | :arrow_down: |
| [...n/tasks/merge\_rollup/MergeRollupTaskGenerator.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUm9sbHVwVGFza0dlbmVyYXRvci5qYXZh) | `94.30% <94.30%> (ø)` | |
| [...gin/minion/tasks/merge\_rollup/MergeProperties.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vfcm9sbHVwL01lcmdlUHJvcGVydGllcy5qYXZh) | `100.00% <100.00%> (ø)` | |
| [...ller/helix/core/minion/TaskTypeMetricsUpdater.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL21pbmlvbi9UYXNrVHlwZU1ldHJpY3NVcGRhdGVyLmphdmE=) | `80.00% <0.00%> (-20.00%)` | :arrow_down: |
| [...impl/dictionary/DoubleOnHeapMutableDictionary.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9yZWFsdGltZS9pbXBsL2RpY3Rpb25hcnkvRG91YmxlT25IZWFwTXV0YWJsZURpY3Rpb25hcnkuamF2YQ==) | `46.98% <0.00%> (-13.26%)` | :arrow_down: |
| [...ore/query/scheduler/resources/ResourceManager.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9xdWVyeS9zY2hlZHVsZXIvcmVzb3VyY2VzL1Jlc291cmNlTWFuYWdlci5qYXZh) | `85.71% <0.00%> (-10.72%)` | :arrow_down: |
| [...mpl/dictionary/DoubleOffHeapMutableDictionary.java](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9yZWFsdGltZS9pbXBsL2RpY3Rpb25hcnkvRG91YmxlT2ZmSGVhcE11dGFibGVEaWN0aW9uYXJ5LmphdmE=) | `51.06% <0.00%> (-7.45%)` | :arrow_down: |
| ... and [25 more](https://codecov.io/gh/apache/incubator-pinot/pull/7178/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [dbaac00...4611426](https://codecov.io/gh/apache/incubator-pinot/pull/7178?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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
[GitHub] [incubator-pinot] snleee commented on pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#issuecomment-882730722
Please fix the test issue. It looks that it has a compilation issue in `MergeRollupTaskGeneratorTest .java`
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r672645646
##########
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()
Review comment:
We use `SegmentLineageBasedSegmentPreSelector.preSelect()` instead of explicitly keeping track of `segmentsNotToMerge`
##########
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()
Review comment:
Computing the entryList is too long and it's hard to understand the logic. Can you compute the entryList first in a separate line and add a bit more comment on what you're trying to compute here?
##########
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);
Review comment:
We should move this line up and put after L124 (after checking skipping conditions)
##########
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
+ if (inCompleteGranularities.containsKey(granularity)) {
+ LOGGER.info("Found incomplete tasks: {} for same granularity: {} and table: {}. Skipping task generation.",
+ inCompleteGranularities.get(granularity), granularity, offlineTableName);
+ continue;
+ }
+
+ MergeProperties mergeProperty = entry.getValue();
+ Set<OfflineSegmentZKMetadata> candidateSegments = new HashSet<>(segmentsForOfflineTable);
+
+ // Get the bucket size and buffer
+ long bucketMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBucketTimePeriod());
+ long bufferMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBufferTimePeriod());
+
+ // Get watermark from MergeRollupTaskMetadata ZNode. WindowStart = watermark. WindowEnd = windowStart + bucket.
+ long windowStartMs = getWatermarkMs(offlineTableName, candidateSegments, bucketMs, granularity);
+ long windowEndMs = windowStartMs + bucketMs;
+
+ // Check that execution window endTimeMs <= now - bufferTime
+ if (windowEndMs > System.currentTimeMillis() - bufferMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} is not older than buffer time: {} configured as {} ago. Skipping task generation: {}",
+ windowStartMs, windowEndMs, bufferMs, bufferMs, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Check that execution window endTimeMs <= waterMark of the lower granularity
+ if (lowerGranularity != null && windowEndMs > lowerGranularityWatermarkMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} of granularity: {} is not older than the watermark: {} of granularity: (). Skipping task generation: {}",
+ windowStartMs, windowEndMs, granularity, lowerGranularityWatermarkMs, lowerGranularity, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Filter out segments that are already merged for the granularity
+ candidateSegments = candidateSegments.stream()
+ .filter(s -> s.getCustomMap() != null ? !granularity.equalsIgnoreCase(
+ s.getCustomMap().get(taskType + MinionConstants.TASK_BUCKET_GRANULARITY_SUFFIX)) : true)
+ .collect(Collectors.toSet());
+
+ // Find segments with data overlapping execution window: windowStart (inclusive) to windowEnd (exclusive),
+ // sort segments based on start time, if start time is the same, sort on end time. Pick segments up to
+ // maxNumRecordsPerTask (soft boundary).
+ List<OfflineSegmentZKMetadata> targetSegments = new ArrayList<>();
+ for (OfflineSegmentZKMetadata offlineSegmentZKMetadata : candidateSegments) {
+ // check overlap with the window
+ if (windowStartMs <= offlineSegmentZKMetadata.getEndTimeMs()
+ && offlineSegmentZKMetadata.getStartTimeMs() < windowEndMs) {
+ targetSegments.add(offlineSegmentZKMetadata);
+ }
+ }
+
+ if (targetSegments.isEmpty()) {
+ LOGGER.info("No segments overlapping with window [{}, {}), Skipping task generation: {}", windowStartMs,
+ windowEndMs, taskType);
+ // Bump up watermark
+ updateWatermarkMs(offlineTableName, granularity, windowStartMs + bucketMs);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs + bucketMs;
+ continue;
+ }
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+
+ Collections.sort(targetSegments, (a, b) -> {
Review comment:
Can you add the comment on what are we sorting based on?
##########
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<>();
Review comment:
The same segment filtering logic is written in `SegmentLineageBasedSegmentPreSelector`
Can we share the filtering logic? We will need to modify slightly the pre-selector as well.
##########
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
+ if (inCompleteGranularities.containsKey(granularity)) {
+ LOGGER.info("Found incomplete tasks: {} for same granularity: {} and table: {}. Skipping task generation.",
+ inCompleteGranularities.get(granularity), granularity, offlineTableName);
+ continue;
+ }
+
+ MergeProperties mergeProperty = entry.getValue();
+ Set<OfflineSegmentZKMetadata> candidateSegments = new HashSet<>(segmentsForOfflineTable);
+
+ // Get the bucket size and buffer
+ long bucketMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBucketTimePeriod());
+ long bufferMs = TimeUtils.convertPeriodToMillis(mergeProperty.getBufferTimePeriod());
+
+ // Get watermark from MergeRollupTaskMetadata ZNode. WindowStart = watermark. WindowEnd = windowStart + bucket.
+ long windowStartMs = getWatermarkMs(offlineTableName, candidateSegments, bucketMs, granularity);
+ long windowEndMs = windowStartMs + bucketMs;
+
+ // Check that execution window endTimeMs <= now - bufferTime
+ if (windowEndMs > System.currentTimeMillis() - bufferMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} is not older than buffer time: {} configured as {} ago. Skipping task generation: {}",
+ windowStartMs, windowEndMs, bufferMs, bufferMs, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Check that execution window endTimeMs <= waterMark of the lower granularity
+ if (lowerGranularity != null && windowEndMs > lowerGranularityWatermarkMs) {
+ LOGGER.info(
+ "Window with start: {} and end: {} of granularity: {} is not older than the watermark: {} of granularity: (). Skipping task generation: {}",
+ windowStartMs, windowEndMs, granularity, lowerGranularityWatermarkMs, lowerGranularity, taskType);
+ lowerGranularity = granularity;
+ lowerGranularityWatermarkMs = windowStartMs;
+ continue;
+ }
+
+ // Filter out segments that are already merged for the granularity
Review comment:
Where do we update the segment zk metadata?
##########
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:
Are we planning to support subtasks (i.e. multiple tasks for a single time bucket using `MAX_NUM_RECORDS_PER_TASK`? If we plan to add it in the next stage, please indicate that as `TODO` comment.
Ideally, we should compute 1 time bucket per granularity and multiple tasks within a bucket is allowed (e.g. splitting a work into multiple tasks for 1 month time bucket)
--
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
[GitHub] [incubator-pinot] snleee commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r673582170
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
There's a valid use case:
1. user pushes data for 07/15
2. pinot merges segments for 07/15
3. user finds out the issue and backfill 07/15 later
4. it's good if pinot merges backfilled segments for 07/15 but we won't do it with the current design since those segments passed the watermark. As we discussed today, one solution is to open up the way for triggering ad-hoc merge-rollup job for a specific window.
--
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
[GitHub] [incubator-pinot] jtao15 commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by 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
[GitHub] [incubator-pinot] Jackie-Jiang commented on a change in pull request #7178: Merge/Rollup task scheduler which supports multi-level CONCAT tasks f…
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on a change in pull request #7178:
URL: https://github.com/apache/incubator-pinot/pull/7178#discussion_r672670245
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -67,16 +68,22 @@ private MinionConstants() {
public static final String GRANULARITY_KEY = "granularity";
// Rollup aggregate function related configs
- public static final String AGGREGATE_KEY_PREFIX = "aggregate";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
// Merge properties related configs
- public static final String MERGE_KEY_PREFIX = "merge";
- public static final String BUFFER_TIME = "bufferTime";
+ public static final String BUFFER_TIME_PERIOD = "bufferTimePeriod";
+ public static final String BUCKET_TIME_PERIOD = "bucketTimePeriod";
+ public static final String ROUND_BUCKET_TIME_PERIOD = "roundBucketTimePeriod";
public static final String MAX_NUM_RECORDS_PER_SEGMENT = "maxNumRecordsPerSegment";
public static final String MAX_NUM_RECORDS_PER_TASK = "maxNumRecordsPerTask";
// Segment name generator related configs
- public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
+ public static final String SEGMENT_NAME_PREFIX_KEY = "segmentNamePrefix";
+ public static final String SEGMENT_NAME_PREFIX = "merged";
+
+ // Window start and window end set by task generator
+ public static final String WINDOW_START_MS_KEY = "windowStartMs";
Review comment:
For merge/rollup task, we should not use the window start/end, but use the partition bucket instead. Different from the realtime-to-offline task, we need to keep all the records (cannot filter the records) for merge/rollup task.
--
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