jtao15 commented on a change in pull request #7481:
URL: https://github.com/apache/pinot/pull/7481#discussion_r726451584
##########
File path:
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGeneratorTest.java
##########
@@ -270,9 +270,121 @@ public void testMaxNumRecordsPerTask() {
.thenReturn(Lists.newArrayList(metadata1, metadata2, metadata3));
pinotTaskConfigs =
generator.generateTasks(Lists.newArrayList(offlineTableConfig));
assertEquals(pinotTaskConfigs.size(), 2);
- checkPinotTaskConfig(pinotTaskConfigs.get(0).getConfigs(), segmentName1 +
"," + segmentName2, "daily", "concat",
+ checkPinotTaskConfig(pinotTaskConfigs.get(0).getConfigs(), segmentName1 +
"," + segmentName2, DAILY, "concat",
"1d", null, "1000000");
- checkPinotTaskConfig(pinotTaskConfigs.get(1).getConfigs(), segmentName3,
"daily", "concat", "1d", null, "1000000");
+ checkPinotTaskConfig(pinotTaskConfigs.get(1).getConfigs(), segmentName3,
DAILY, "concat", "1d", null, "1000000");
+ }
+
+ /**
+ * Test num parallel buckets
+ */
+ @Test
+ public void testNumParallelBuckets() {
+ Map<String, Map<String, String>> taskConfigsMap = new HashMap<>();
+ Map<String, String> tableTaskConfigs = new HashMap<>();
+ tableTaskConfigs.put("daily.mergeType", "concat");
+ tableTaskConfigs.put("daily.bufferTimePeriod", "2d");
+ tableTaskConfigs.put("daily.bucketTimePeriod", "1d");
+ tableTaskConfigs.put("daily.numParallelBuckets", "3");
+ tableTaskConfigs.put("daily.maxNumRecordsPerSegment", "1000000");
+ taskConfigsMap.put(MinionConstants.MergeRollupTask.TASK_TYPE,
tableTaskConfigs);
+ TableConfig offlineTableConfig = getOfflineTableConfig(taskConfigsMap);
+ ClusterInfoAccessor mockClusterInfoProvide =
mock(ClusterInfoAccessor.class);
+
+ String segmentName1 = "testTable__1";
+ String segmentName2 = "testTable__2";
+ String segmentName3 = "testTable__3";
+ String segmentName4 = "testTable__4";
+ String segmentName5 = "testTable__5";
+ SegmentZKMetadata metadata1 =
+ getSegmentZKMetadata(segmentName1, 86_400_000L, 90_000_000L,
TimeUnit.MILLISECONDS, "download1");
+ SegmentZKMetadata metadata2 =
+ getSegmentZKMetadata(segmentName2, 86_400_000L, 100_000_000L,
TimeUnit.MILLISECONDS, "download2");
+ SegmentZKMetadata metadata3 =
+ getSegmentZKMetadata(segmentName3, 172_800_000L, 173_000_000L,
TimeUnit.MILLISECONDS, "download3");
+ SegmentZKMetadata metadata4 =
+ getSegmentZKMetadata(segmentName4, 259_200_000L, 260_000_000L,
TimeUnit.MILLISECONDS, "download4");
+ SegmentZKMetadata metadata5 =
+ getSegmentZKMetadata(segmentName5, 345_600_000L, 346_000_000L,
TimeUnit.MILLISECONDS, "download5");
+
+ // No spilled over data
+ when(mockClusterInfoProvide.getSegmentsZKMetadata(OFFLINE_TABLE_NAME))
+ .thenReturn(Lists.newArrayList(metadata1, metadata2, metadata3,
metadata4, metadata5));
+ MergeRollupTaskGenerator generator = new MergeRollupTaskGenerator();
+ generator.init(mockClusterInfoProvide);
+ List<PinotTaskConfig> pinotTaskConfigs =
generator.generateTasks(Lists.newArrayList(offlineTableConfig));
+ assertEquals(pinotTaskConfigs.size(), 3);
+ checkPinotTaskConfig(pinotTaskConfigs.get(0).getConfigs(), segmentName1 +
"," + segmentName2, DAILY, "concat",
+ "1d", null, "1000000");
+ checkPinotTaskConfig(pinotTaskConfigs.get(1).getConfigs(), segmentName3,
DAILY, "concat",
+ "1d", null, "1000000");
+ checkPinotTaskConfig(pinotTaskConfigs.get(2).getConfigs(), segmentName4,
DAILY, "concat",
+ "1d", null, "1000000");
+
+ // Has spilled over data
+ String segmentName6 = "testTable__6";
+ SegmentZKMetadata metadata6 =
+ getSegmentZKMetadata(segmentName6, 172_800_000L, 260_000_000L,
TimeUnit.MILLISECONDS, null);
+ when(mockClusterInfoProvide.getSegmentsZKMetadata(OFFLINE_TABLE_NAME))
+ .thenReturn(Lists.newArrayList(metadata1, metadata2, metadata3,
metadata4, metadata5, metadata6));
+ pinotTaskConfigs =
generator.generateTasks(Lists.newArrayList(offlineTableConfig));
+ assertEquals(pinotTaskConfigs.size(), 2);
+ checkPinotTaskConfig(pinotTaskConfigs.get(0).getConfigs(), segmentName1 +
"," + segmentName2, DAILY, "concat",
+ "1d", null, "1000000");
+ checkPinotTaskConfig(pinotTaskConfigs.get(1).getConfigs(), segmentName3 +
"," + segmentName6, DAILY, "concat",
+ "1d", null, "1000000");
+
+ // Has time bucket without overlapping segments
+ when(mockClusterInfoProvide.getSegmentsZKMetadata(OFFLINE_TABLE_NAME))
+ .thenReturn(Lists.newArrayList(metadata1, metadata2, metadata4,
metadata5));
+ pinotTaskConfigs =
generator.generateTasks(Lists.newArrayList(offlineTableConfig));
+ assertEquals(pinotTaskConfigs.size(), 3);
+ checkPinotTaskConfig(pinotTaskConfigs.get(0).getConfigs(), segmentName1 +
"," + segmentName2, DAILY, "concat",
+ "1d", null, "1000000");
+ checkPinotTaskConfig(pinotTaskConfigs.get(1).getConfigs(), segmentName4,
DAILY, "concat",
+ "1d", null, "1000000");
+ checkPinotTaskConfig(pinotTaskConfigs.get(2).getConfigs(), segmentName5,
DAILY, "concat",
+ "1d", null, "1000000");
+
+ // Heading buckets are merged
Review comment:
Yes, updated.
--
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-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
