Jackie-Jiang commented on code in PR #8589:
URL: https://github.com/apache/pinot/pull/8589#discussion_r882186480
##########
pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java:
##########
@@ -158,6 +159,17 @@ public void start()
Utils.logVersions();
MinionContext minionContext = MinionContext.getInstance();
+ SegmentPurger.RecordPurgerFactory recordPurgerFactory = new
SegmentPurger.RecordPurgerFactory() {
Review Comment:
(MAJOR) This can break existing starter wrapper if the record purger factory
is set before calling `start()`. Let's not put a default record purger factory.
If purge task is enabled, user should have a starter wrapper to plug-in their
own custom record purger factory
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.purge;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.pinot.common.data.Segment;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.controller.api.exception.UnknownTaskTypeException;
+import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator;
+import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+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;
+
+
+@TaskGenerator
+public class PurgeTaskGenerator extends BaseTaskGenerator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PurgeTaskGenerator.class);
+ private static final String DEFAULT_DELTA_PERIOD = "1d";
Review Comment:
Move this to the `PurgeTask` constant
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.purge;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.pinot.common.data.Segment;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.controller.api.exception.UnknownTaskTypeException;
+import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator;
+import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+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;
+
+
+@TaskGenerator
+public class PurgeTaskGenerator extends BaseTaskGenerator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PurgeTaskGenerator.class);
+ private static final String DEFAULT_DELTA_PERIOD = "1d";
+
+ @Override
+ public String getTaskType() {
+ return MinionConstants.PurgeTask.TASK_TYPE;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+ LOGGER.info("Start generating PurgeTask");
+ String taskType = MinionConstants.PurgeTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+
+ for (TableConfig tableConfig : tableConfigs) {
+
+ String tableName = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ LOGGER.info("Task type : {}, cannot be run on table of type {}",
taskType, TableType.REALTIME);
Review Comment:
(minor)
```suggestion
LOGGER.warn("Skip generating task: {} for real-time table: {}",
taskType, tableName);
```
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.purge;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.pinot.common.data.Segment;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.controller.api.exception.UnknownTaskTypeException;
+import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator;
+import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+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;
+
+
+@TaskGenerator
+public class PurgeTaskGenerator extends BaseTaskGenerator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PurgeTaskGenerator.class);
+ private static final String DEFAULT_DELTA_PERIOD = "1d";
+
+ @Override
+ public String getTaskType() {
+ return MinionConstants.PurgeTask.TASK_TYPE;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+ LOGGER.info("Start generating PurgeTask");
+ String taskType = MinionConstants.PurgeTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+
+ for (TableConfig tableConfig : tableConfigs) {
+
+ String tableName = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ LOGGER.info("Task type : {}, cannot be run on table of type {}",
taskType, TableType.REALTIME);
+ continue;
+ }
+
+ Map<String, String> taskConfigs;
+ try {
+ TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
+ Preconditions.checkNotNull(tableTaskConfig);
+ taskConfigs =
tableTaskConfig.getConfigsForTaskType(MinionConstants.PurgeTask.TASK_TYPE);
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null
for Table: {}", tableName);
+ } catch (Exception e) {
+ continue;
+ }
+ String deltaTimePeriod =
+
taskConfigs.getOrDefault(MinionConstants.PurgeTask.DELTA_TIME_PERIOD_KEY,
DEFAULT_DELTA_PERIOD);
+ long purgeDeltaMs = TimeUtils.convertPeriodToMillis(deltaTimePeriod);
+
+ LOGGER.info("Start generating task configs for table: {} for task: {}",
tableName, taskType);
+ // Get max number of tasks for this table
+ int tableMaxNumTasks;
+ String tableMaxNumTasksConfig =
taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
+ if (tableMaxNumTasksConfig != null) {
+ try {
+ tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
+ } catch (Exception e) {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and
task {}", tableName, taskType);
+ }
+ } else {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ }
+ List<SegmentZKMetadata> offlineSegmentsZKMetadata =
_clusterInfoAccessor.getSegmentsZKMetadata(tableName);
+
+ Predicate<SegmentZKMetadata> alreadyPurged = segment ->
segment.getCustomMap() != null;
+ Predicate<SegmentZKMetadata> notPurged = segment ->
segment.getCustomMap() == null;
+
+ List<SegmentZKMetadata> purgedSegmentsZKMetadata =
Review Comment:
(minor) Suggest using regular for loops for better performance
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.purge;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.pinot.common.data.Segment;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.controller.api.exception.UnknownTaskTypeException;
+import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator;
+import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+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;
+
+
+@TaskGenerator
+public class PurgeTaskGenerator extends BaseTaskGenerator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PurgeTaskGenerator.class);
+ private static final String DEFAULT_DELTA_PERIOD = "1d";
+
+ @Override
+ public String getTaskType() {
+ return MinionConstants.PurgeTask.TASK_TYPE;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+ LOGGER.info("Start generating PurgeTask");
+ String taskType = MinionConstants.PurgeTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+
+ for (TableConfig tableConfig : tableConfigs) {
+
+ String tableName = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ LOGGER.info("Task type : {}, cannot be run on table of type {}",
taskType, TableType.REALTIME);
+ continue;
+ }
+
+ Map<String, String> taskConfigs;
+ try {
+ TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
+ Preconditions.checkNotNull(tableTaskConfig);
Review Comment:
Let's not do pre-condition and try-catch. We can do if check then log
warning if it fails the check:
```suggestion
if (tableTaskConfig == null) {
LOGGER.warn("Failed to find task config for table: {}", tableName);
continue;
}
...
```
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.purge;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.pinot.common.data.Segment;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.controller.api.exception.UnknownTaskTypeException;
+import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator;
+import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+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;
+
+
+@TaskGenerator
+public class PurgeTaskGenerator extends BaseTaskGenerator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PurgeTaskGenerator.class);
+ private static final String DEFAULT_DELTA_PERIOD = "1d";
+
+ @Override
+ public String getTaskType() {
+ return MinionConstants.PurgeTask.TASK_TYPE;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+ LOGGER.info("Start generating PurgeTask");
+ String taskType = MinionConstants.PurgeTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+
+ for (TableConfig tableConfig : tableConfigs) {
+
+ String tableName = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ LOGGER.info("Task type : {}, cannot be run on table of type {}",
taskType, TableType.REALTIME);
+ continue;
+ }
+
+ Map<String, String> taskConfigs;
+ try {
+ TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
+ Preconditions.checkNotNull(tableTaskConfig);
+ taskConfigs =
tableTaskConfig.getConfigsForTaskType(MinionConstants.PurgeTask.TASK_TYPE);
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null
for Table: {}", tableName);
+ } catch (Exception e) {
+ continue;
+ }
+ String deltaTimePeriod =
+
taskConfigs.getOrDefault(MinionConstants.PurgeTask.DELTA_TIME_PERIOD_KEY,
DEFAULT_DELTA_PERIOD);
+ long purgeDeltaMs = TimeUtils.convertPeriodToMillis(deltaTimePeriod);
+
+ LOGGER.info("Start generating task configs for table: {} for task: {}",
tableName, taskType);
+ // Get max number of tasks for this table
+ int tableMaxNumTasks;
+ String tableMaxNumTasksConfig =
taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
+ if (tableMaxNumTasksConfig != null) {
+ try {
+ tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
+ } catch (Exception e) {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and
task {}", tableName, taskType);
+ }
+ } else {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ }
+ List<SegmentZKMetadata> offlineSegmentsZKMetadata =
_clusterInfoAccessor.getSegmentsZKMetadata(tableName);
+
+ Predicate<SegmentZKMetadata> alreadyPurged = segment ->
segment.getCustomMap() != null;
+ Predicate<SegmentZKMetadata> notPurged = segment ->
segment.getCustomMap() == null;
+
+ List<SegmentZKMetadata> purgedSegmentsZKMetadata =
+
offlineSegmentsZKMetadata.stream().filter(alreadyPurged).collect(Collectors.toList());
+
+ List<SegmentZKMetadata> notpurgedSegmentsZKMetadata =
+
offlineSegmentsZKMetadata.stream().filter(notPurged).collect(Collectors.toList());
+
+ Collections.sort(purgedSegmentsZKMetadata, Comparator.comparing(
+ segmentZKMetadata -> segmentZKMetadata.getCustomMap()
+ .get(MinionConstants.PurgeTask.TASK_TYPE +
MinionConstants.TASK_TIME_SUFFIX),
+ Comparator.nullsFirst(Comparator.naturalOrder())));
+ //add already purged segment at the end
+ notpurgedSegmentsZKMetadata.addAll(purgedSegmentsZKMetadata);
+
+ int tableNumTasks = 0;
+ Set<Segment> runningSegments =
+
TaskGeneratorUtils.getRunningSegments(MinionConstants.PurgeTask.TASK_TYPE,
_clusterInfoAccessor);
+ for (SegmentZKMetadata segmentZKMetadata : notpurgedSegmentsZKMetadata) {
+ Map<String, String> configs = new HashMap<>();
+ String segmentName = segmentZKMetadata.getSegmentName();
+ Long tsLastPurge;
+ if (segmentZKMetadata.getCustomMap() != null) {
+ tsLastPurge = Long.valueOf(segmentZKMetadata.getCustomMap()
+ .get(MinionConstants.PurgeTask.TASK_TYPE +
MinionConstants.TASK_TIME_SUFFIX));
+ } else {
+ tsLastPurge = 0L;
+ }
+
+ //skip running segment
+ if (runningSegments.contains(new Segment(tableName, segmentName))) {
+ continue;
+ }
+ if ((tsLastPurge != null) && ((System.currentTimeMillis() -
tsLastPurge) < purgeDeltaMs)) {
+ //skip if purge delay is not reached
+ continue;
+ }
+ if (tableNumTasks == tableMaxNumTasks) {
+ break;
+ }
+ configs.put(MinionConstants.TABLE_NAME_KEY, tableName);
+ configs.put(MinionConstants.SEGMENT_NAME_KEY, segmentName);
+ configs.put(MinionConstants.DOWNLOAD_URL_KEY,
segmentZKMetadata.getDownloadUrl());
+ configs.put(MinionConstants.UPLOAD_URL_KEY,
_clusterInfoAccessor.getVipUrl() + "/segments");
+ configs.put(MinionConstants.ORIGINAL_SEGMENT_CRC_KEY,
String.valueOf(segmentZKMetadata.getCrc()));
+ pinotTaskConfigs.add(new PinotTaskConfig(taskType, configs));
+ tableNumTasks++;
+ }
+ LOGGER.info("Finished generating {} tasks configs for table: {} " + "for
task: {}", tableNumTasks, tableName,
+ taskType);
+ }
+ return pinotTaskConfigs;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(TableConfig tableConfig,
Map<String, String> taskConfigs)
Review Comment:
(minor) No need to override since it is the same as the base implementation
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.purge;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.pinot.common.data.Segment;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.controller.api.exception.UnknownTaskTypeException;
+import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator;
+import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+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;
+
+
+@TaskGenerator
+public class PurgeTaskGenerator extends BaseTaskGenerator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PurgeTaskGenerator.class);
+ private static final String DEFAULT_DELTA_PERIOD = "1d";
+
+ @Override
+ public String getTaskType() {
+ return MinionConstants.PurgeTask.TASK_TYPE;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+ LOGGER.info("Start generating PurgeTask");
+ String taskType = MinionConstants.PurgeTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+
+ for (TableConfig tableConfig : tableConfigs) {
+
+ String tableName = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ LOGGER.info("Task type : {}, cannot be run on table of type {}",
taskType, TableType.REALTIME);
+ continue;
+ }
+
+ Map<String, String> taskConfigs;
+ try {
+ TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
+ Preconditions.checkNotNull(tableTaskConfig);
+ taskConfigs =
tableTaskConfig.getConfigsForTaskType(MinionConstants.PurgeTask.TASK_TYPE);
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null
for Table: {}", tableName);
+ } catch (Exception e) {
+ continue;
+ }
+ String deltaTimePeriod =
+
taskConfigs.getOrDefault(MinionConstants.PurgeTask.DELTA_TIME_PERIOD_KEY,
DEFAULT_DELTA_PERIOD);
+ long purgeDeltaMs = TimeUtils.convertPeriodToMillis(deltaTimePeriod);
+
+ LOGGER.info("Start generating task configs for table: {} for task: {}",
tableName, taskType);
+ // Get max number of tasks for this table
+ int tableMaxNumTasks;
+ String tableMaxNumTasksConfig =
taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
+ if (tableMaxNumTasksConfig != null) {
+ try {
+ tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
+ } catch (Exception e) {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and
task {}", tableName, taskType);
+ }
+ } else {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ }
+ List<SegmentZKMetadata> offlineSegmentsZKMetadata =
_clusterInfoAccessor.getSegmentsZKMetadata(tableName);
+
+ Predicate<SegmentZKMetadata> alreadyPurged = segment ->
segment.getCustomMap() != null;
+ Predicate<SegmentZKMetadata> notPurged = segment ->
segment.getCustomMap() == null;
+
+ List<SegmentZKMetadata> purgedSegmentsZKMetadata =
+
offlineSegmentsZKMetadata.stream().filter(alreadyPurged).collect(Collectors.toList());
+
+ List<SegmentZKMetadata> notpurgedSegmentsZKMetadata =
+
offlineSegmentsZKMetadata.stream().filter(notPurged).collect(Collectors.toList());
+
+ Collections.sort(purgedSegmentsZKMetadata, Comparator.comparing(
+ segmentZKMetadata -> segmentZKMetadata.getCustomMap()
+ .get(MinionConstants.PurgeTask.TASK_TYPE +
MinionConstants.TASK_TIME_SUFFIX),
+ Comparator.nullsFirst(Comparator.naturalOrder())));
+ //add already purged segment at the end
+ notpurgedSegmentsZKMetadata.addAll(purgedSegmentsZKMetadata);
Review Comment:
We can avoid reading the same key multiple times by first processing the
last purge time and put it alone with the segment metadata
(`Pair<SegmentZKMetadata, Long>`). If the segment is never purged, put 0.
##########
pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java:
##########
@@ -67,6 +67,7 @@ public static class ConvertToRawIndexTask {
// Purges rows inside segment that match chosen criteria
public static class PurgeTask {
public static final String TASK_TYPE = "PurgeTask";
+ public static final String DELTA_TIME_PERIOD_KEY = "deltaPurgeTimePeriod";
Review Comment:
Checked with @jackjlli, we may change it as followings
```suggestion
public static final String LAST_PURGE_TIME_THRESHOLD_PERIOD =
"lastPurgeTimeThresholdPeriod";
public static final String DEFAULT_LAST_PURGE_TIME_THRESHOLD_PERIOD =
"14d";
```
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.purge;
+
+import com.google.common.base.Preconditions;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.apache.pinot.common.data.Segment;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.controller.api.exception.UnknownTaskTypeException;
+import
org.apache.pinot.controller.helix.core.minion.generator.BaseTaskGenerator;
+import
org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorUtils;
+import org.apache.pinot.core.common.MinionConstants;
+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;
+
+
+@TaskGenerator
+public class PurgeTaskGenerator extends BaseTaskGenerator {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(PurgeTaskGenerator.class);
+ private static final String DEFAULT_DELTA_PERIOD = "1d";
+
+ @Override
+ public String getTaskType() {
+ return MinionConstants.PurgeTask.TASK_TYPE;
+ }
+
+ @Override
+ public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+ LOGGER.info("Start generating PurgeTask");
+ String taskType = MinionConstants.PurgeTask.TASK_TYPE;
+ List<PinotTaskConfig> pinotTaskConfigs = new ArrayList<>();
+
+ for (TableConfig tableConfig : tableConfigs) {
+
+ String tableName = tableConfig.getTableName();
+ if (tableConfig.getTableType() == TableType.REALTIME) {
+ LOGGER.info("Task type : {}, cannot be run on table of type {}",
taskType, TableType.REALTIME);
+ continue;
+ }
+
+ Map<String, String> taskConfigs;
+ try {
+ TableTaskConfig tableTaskConfig = tableConfig.getTaskConfig();
+ Preconditions.checkNotNull(tableTaskConfig);
+ taskConfigs =
tableTaskConfig.getConfigsForTaskType(MinionConstants.PurgeTask.TASK_TYPE);
+ Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null
for Table: {}", tableName);
+ } catch (Exception e) {
+ continue;
+ }
+ String deltaTimePeriod =
+
taskConfigs.getOrDefault(MinionConstants.PurgeTask.DELTA_TIME_PERIOD_KEY,
DEFAULT_DELTA_PERIOD);
+ long purgeDeltaMs = TimeUtils.convertPeriodToMillis(deltaTimePeriod);
+
+ LOGGER.info("Start generating task configs for table: {} for task: {}",
tableName, taskType);
+ // Get max number of tasks for this table
+ int tableMaxNumTasks;
+ String tableMaxNumTasksConfig =
taskConfigs.get(MinionConstants.TABLE_MAX_NUM_TASKS_KEY);
+ if (tableMaxNumTasksConfig != null) {
+ try {
+ tableMaxNumTasks = Integer.parseInt(tableMaxNumTasksConfig);
+ } catch (Exception e) {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ LOGGER.warn("MaxNumTasks have been wrongly set for table : {}, and
task {}", tableName, taskType);
+ }
+ } else {
+ tableMaxNumTasks = Integer.MAX_VALUE;
+ }
+ List<SegmentZKMetadata> offlineSegmentsZKMetadata =
_clusterInfoAccessor.getSegmentsZKMetadata(tableName);
+
+ Predicate<SegmentZKMetadata> alreadyPurged = segment ->
segment.getCustomMap() != null;
Review Comment:
custom map existence doesn't mean the segment is already purged. We need to
check if the custom map contains key `MinionConstants.PurgeTask.TASK_TYPE +
MinionConstants.TASK_TIME_SUFFIX`
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
