Implement builtin-purge task #8589

Airliquide76 · 2022-04-25T14:29:21Z

Instructions:

Feature to facilitate the purge process.

Tested mannually with my own use case not shared here. As I've developped the logic of wich record to purge directly in the BaseMinionStarter.

The RecordPurgerFactory implementation should be the only thing to do for a user to purge records.

jackjlli · 2022-04-25T23:04:49Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+
+    private static final PurgeTaskGenerator INSTANCE = new PurgeTaskGenerator();
+
+    protected ClusterInfoAccessor _clusterInfoAccessor;


Change it to private?

jackjlli · 2022-04-25T23:11:07Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+
+@TaskGenerator
+public class PurgeTaskGenerator implements PinotTaskGenerator {
+    protected static final Logger LOGGER = LoggerFactory.getLogger(PurgeTaskGenerator.class);


Same here. Change it to private.

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

jackjlli · 2022-04-25T23:14:10Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+
+    @Override
+    public void init(ClusterInfoAccessor clusterInfoAccessor) {
+        LOGGER.info("I'm in init for the purge");


Update the message to sth like Initializing purge task?

Woops -> fixed in comming commit

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

jackjlli · 2022-04-25T23:19:36Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+            Preconditions.checkNotNull(tableTaskConfig);
+            Map<String, String> taskConfigs =
+                    tableTaskConfig.getConfigsForTaskType(MinionConstants.PurgeTask.TASK_TYPE);
+            Preconditions.checkNotNull(taskConfigs, "Task config shouldn't be null for Table: {}", tableName);


If one table fails on this precondition, the whole purge task generation for other tables will be stopped. Maybe we should continue with other tables?

Ok. I've surrounded with try catch to continue loop if Preconditions.checkNotNull for the task throw NullPointerException

jackjlli · 2022-04-25T23:20:18Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+                        taskType, TableType.REALTIME);
+                continue;
+            }
+                // Get max number of tasks for this table


Please align with the indentation

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

Airliquide76 · 2022-04-26T09:16:50Z

There is a git issue somewhere on my side I've added only a single commit and other I don't even know pops into this PR 👎 I will check this

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

Airliquide76 · 2022-04-28T08:02:06Z

Ok I've fix all the reviews on this. Many thanks for you patience @jackjlli 👍

jackjlli

Please add one TODO in the class that the recently purged segment should be scheduled at last so that it won't be scheduled again and again, whereas other segments don't have the chance to be scheduled.

...uiltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskExecutor.java

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

Jackie-Jiang · 2022-05-03T18:46:40Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+
+    @Override
+    public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
+        // TODO: Pick lastly purged segment first to avoid looping again and again on the same segment


The purged segments should have a purge timestamp in the segment ZK metadata custom fields. We should rely on that to order the purged segments, and add a threshold and not purge the recently purged segments again and again

Yes for sure.
I've added here a (set/get)PurgeTime in SegmentZKMetadata but as I've implemented timestamp will be set event if task is failed 👎 Not a super fan of this thing. Not pushed yet because I can't figure out how tu push the segment metatda from the TaskGenerator. Any example ?

Let's not add new field into the SegmentZKMetadata. The purge time is already available in the custom field of the SegmentZKMetadata under the key PurgeTask.time. You may take a look at the PurgeTaskExecutor class for more context

Nice that solve most of my issues :) I've pushed a first version null safe using comparator to sort the array of segments. I will look arround to other taskGenerators to see how implement the threesold

Airliquide76 · 2022-05-18T09:16:17Z

Hi made a few change based on you review (many thanks for the improvement made to my original naive implementation). Is that possible to continue the review with the recent change made @Jackie-Jiang or @jackjlli ? Thank you.

Jackie-Jiang

Can you please add an integration test to ensure the purge task works end-to-end? You may refer to SimpleMinionClusterIntegrationTest and MergeRollupMinionClusterIntegrationTest

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

Jackie-Jiang · 2022-05-18T21:43:17Z

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


@jackjlli Can you please check the key used in LinkedIn? We want to make them the same so that you can eventually move to the open source implementation without causing backward incompatible issue

Integration test added. Was hard to manipulate segmentMetadata in the setUp. I also was forced to implement a purge (does not purge return false every row and a BIG TODO in the baseMinionStarter ) to run the tests.

Open to review again @Jackie-Jiang many thx for you time

Checked with @jackjlli, we may change it as followings

Suggested change

public static final String DELTA_TIME_PERIOD_KEY = "deltaPurgeTimePeriod";

public static final String LAST_PURGE_TIME_THRESHOLD_PERIOD = "lastPurgeTimeThresholdPeriod";

public static final String DEFAULT_LAST_PURGE_TIME_THRESHOLD_PERIOD = "14d";

codecov-commenter · 2022-05-21T07:47:14Z

Codecov Report

Merging #8589 (5bf84c8) into master (4d36f3d) will increase coverage by 40.23%.
The diff coverage is 76.81%.

@@              Coverage Diff              @@
##             master    #8589       +/-   ##
=============================================
+ Coverage     29.41%   69.64%   +40.23%     
- Complexity        0     4617     +4617     
=============================================
  Files          1680     1736       +56     
  Lines         88398    91279     +2881     
  Branches      13432    13641      +209     
=============================================
+ Hits          25998    63574    +37576     
+ Misses        60013    23276    -36737     
- Partials       2387     4429     +2042

Flag	Coverage Δ
integration1	`26.90% <4.34%> (-0.37%)`	⬇️
integration2	`25.62% <76.81%> (+0.06%)`	⬆️
unittests1	`66.15% <ø> (?)`
unittests2	`14.13% <4.34%> (?)`

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
.../org/apache/pinot/core/common/MinionConstants.java	`0.00% <ø> (ø)`
.../plugin/minion/tasks/purge/PurgeTaskGenerator.java	`76.81% <76.81%> (ø)`
...ava/org/apache/pinot/core/auth/BasicAuthUtils.java	`28.81% <0.00%> (-48.46%)`	⬇️
...che/pinot/controller/util/TableMetadataReader.java	`63.41% <0.00%> (-36.59%)`	⬇️
...ache/pinot/common/metadata/ZKMetadataProvider.java	`65.92% <0.00%> (-13.28%)`	⬇️
...y/aggregation/function/SumAggregationFunction.java	`84.00% <0.00%> (-11.46%)`	⬇️
...g/apache/pinot/sql/parsers/CalciteSqlCompiler.java	`88.88% <0.00%> (-11.12%)`	⬇️
.../org/apache/pinot/core/util/QueryOptionsUtils.java	`65.00% <0.00%> (-7.23%)`	⬇️
...nction/DistinctCountBitmapAggregationFunction.java	`47.66% <0.00%> (-6.22%)`	⬇️
...inot/controller/api/access/AccessControlUtils.java	`73.91% <0.00%> (-6.09%)`	⬇️
... and 1365 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 4d36f3d...5bf84c8. Read the comment docs.

Airliquide76 · 2022-05-23T19:23:38Z

So what is the next step ? 😄

Jackie-Jiang

@jackjlli @siddharthteotia Can you please take a look at the new config key DELTA_TIME_PERIOD_KEY? I'd suggest making it the same as the one used in LI so that in the future you may migrate to this one without compatible issue

...ests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java

pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

jackjlli · 2022-05-24T00:04:08Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+      } catch (Exception e) {
+        continue;
+      }
+      String deltaTimePeriod =


You don't need to introduce a new concept of DELTA_TIME_PERIOD_KEY to schedule the purge task. Instead the purge task should be set up on segment level. And since the number of segments is different for different tables, this new config will schedule unbalanced purge tasks per day. So I suggest removing this new table level DELTA_TIME_PERIOD_KEY.
Some variable called lastPurgeTime can be calculated for each of the segments. By default the value should be the segment creation time. If there is a field called "PurgeTask.time" in the custom map in the ZK metadata, it should override the default value.
You can leave a TODO in this generator, and I'll fill the logic once this PR is merged.

I've implement the threesold asked here : #8589 (comment) .

@jackjlli This delta is essentially the time threshold to skip purging if the segment is just purged within the threshold (e.g. 2 weeks for GDPR requirement)

Airliquide76 · 2022-05-24T13:16:56Z

Review fixed. I'm now waiting for the DELTA_TIME_PERIOD_KEY value from linkedin team to integrate this thing without breaking what have been done on their side.

Jackie-Jiang · 2022-05-25T22:59:03Z

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


Checked with @jackjlli, we may change it as followings

Suggested change

public static final String DELTA_TIME_PERIOD_KEY = "deltaPurgeTimePeriod";

public static final String LAST_PURGE_TIME_THRESHOLD_PERIOD = "lastPurgeTimeThresholdPeriod";

public static final String DEFAULT_LAST_PURGE_TIME_THRESHOLD_PERIOD = "14d";

pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

Jackie-Jiang · 2022-05-25T23:18:45Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+              .get(MinionConstants.PurgeTask.TASK_TYPE + MinionConstants.TASK_TIME_SUFFIX),
+          Comparator.nullsFirst(Comparator.naturalOrder())));
+      //add already purged segment at the end
+      notpurgedSegmentsZKMetadata.addAll(purgedSegmentsZKMetadata);


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.

yes but SegmentZKMetadata should then implement Serializable. Not really aware of all the impacts ...

We can use org.apache.commons.lang3.tuple.Pair which does not require key to be Serializable

Jackie-Jiang

Mostly good, some minor comments

Jackie-Jiang · 2022-05-27T22:19:14Z

...iltin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/purge/PurgeTaskGenerator.java

+              .get(MinionConstants.PurgeTask.TASK_TYPE + MinionConstants.TASK_TIME_SUFFIX),
+          Comparator.nullsFirst(Comparator.naturalOrder())));
+      //add already purged segment at the end
+      notpurgedSegmentsZKMetadata.addAll(purgedSegmentsZKMetadata);


We can use org.apache.commons.lang3.tuple.Pair which does not require key to be Serializable

...ests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java

Airliquide76 · 2022-05-30T08:30:12Z

Well I've fixed the minors review except the Pair thing. I don't know why the CI have failed on the integration tests as they all run well on my machine 👎

jackjlli

LGTM. Thanks for adding the generator! 👍

Airliquide76 mentioned this pull request Apr 25, 2022

Implement purge process #8588

Closed

jackjlli reviewed Apr 25, 2022

View reviewed changes

Airliquide76 closed this Apr 26, 2022

Airliquide76 force-pushed the PurgeImplementatation branch from a865605 to 4d36f3d Compare April 26, 2022 12:17

francois autaa added 3 commits April 26, 2022 14:18

Implement builtin-purge task

e541bd0

Fix PR review v0.1

3f715e3

Fix checkstyle

ca44a41

Airliquide76 reopened this Apr 26, 2022

jackjlli reviewed Apr 26, 2022

View reviewed changes

Fix PR review v0.2

eb87edf

jackjlli reviewed Apr 28, 2022

View reviewed changes

francois autaa added 2 commits April 29, 2022 09:32

Add todo to purge lastly purged segment first

f64a46d

Fix PR review V0.3

d54ab0b

Jackie-Jiang reviewed May 3, 2022

View reviewed changes

francois autaa added 2 commits May 5, 2022 09:53

Add sergment purging order

efad166

add timeIntervalCheck for purge task

835ea6e

Jackie-Jiang reviewed May 18, 2022

View reviewed changes

francois autaa added 2 commits May 20, 2022 11:53

add integration test

2be8cbc

add integration test

f43ff06

Jackie-Jiang reviewed May 23, 2022

View reviewed changes

jackjlli reviewed May 23, 2022

View reviewed changes

jackjlli reviewed May 24, 2022

View reviewed changes

Fix PR reviews

4c59ea3

Airliquide76 requested a review from Jackie-Jiang May 25, 2022 08:57

Airliquide76 requested a review from jackjlli May 25, 2022 08:57

Jackie-Jiang reviewed May 25, 2022

View reviewed changes

francois autaa added 2 commits May 27, 2022 15:16

Fix PR review

3802d58

Fix PR sout

3a5a4b7

Jackie-Jiang approved these changes May 27, 2022

View reviewed changes

Fix PR minor issues

5bf84c8

jackjlli approved these changes May 31, 2022

View reviewed changes

jackjlli merged commit d2fa9dd into apache:master May 31, 2022

Airliquide76 deleted the PurgeImplementatation branch May 31, 2022 07:05


		private static final PurgeTaskGenerator INSTANCE = new PurgeTaskGenerator();

		protected ClusterInfoAccessor _clusterInfoAccessor;

	public static final String DELTA_TIME_PERIOD_KEY = "deltaPurgeTimePeriod";
	public static final String LAST_PURGE_TIME_THRESHOLD_PERIOD = "lastPurgeTimeThresholdPeriod";
	public static final String DEFAULT_LAST_PURGE_TIME_THRESHOLD_PERIOD = "14d";

Implement builtin-purge task #8589

Implement builtin-purge task #8589

Conversation

Airliquide76 commented Apr 25, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Airliquide76 commented Apr 26, 2022

Airliquide76 commented Apr 28, 2022

jackjlli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Airliquide76 May 5, 2022 • edited Loading

Choose a reason for hiding this comment

Airliquide76 commented May 18, 2022

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented May 21, 2022 • edited Loading

Codecov Report

Airliquide76 commented May 23, 2022

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Airliquide76 commented May 24, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Airliquide76 commented May 30, 2022

jackjlli left a comment

Choose a reason for hiding this comment

Airliquide76 May 5, 2022 •

edited

Loading

codecov-commenter commented May 21, 2022 •

edited

Loading