| /** |
| * 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.integration.tests; |
| |
| import com.fasterxml.jackson.databind.JsonNode; |
| import java.io.File; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import javax.annotation.Nullable; |
| import org.apache.commons.io.FileUtils; |
| import org.apache.helix.task.TaskState; |
| import org.apache.helix.zookeeper.datamodel.ZNRecord; |
| import org.apache.pinot.common.lineage.SegmentLineageAccessHelper; |
| import org.apache.pinot.common.metadata.segment.SegmentZKMetadata; |
| import org.apache.pinot.common.metrics.MetricValueUtils; |
| import org.apache.pinot.common.minion.MergeRollupTaskMetadata; |
| import org.apache.pinot.common.minion.MinionTaskMetadataUtils; |
| import org.apache.pinot.common.utils.SqlResultComparator; |
| import org.apache.pinot.common.utils.TarGzCompressionUtils; |
| import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; |
| import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager; |
| import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; |
| import org.apache.pinot.core.common.MinionConstants; |
| import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; |
| import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; |
| import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver; |
| import org.apache.pinot.spi.config.table.ColumnPartitionConfig; |
| import org.apache.pinot.spi.config.table.SegmentPartitionConfig; |
| 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.data.Schema; |
| import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties; |
| import org.apache.pinot.spi.stream.StreamConfigProperties; |
| import org.apache.pinot.spi.utils.builder.TableConfigBuilder; |
| import org.apache.pinot.spi.utils.builder.TableNameBuilder; |
| import org.apache.pinot.tools.utils.KafkaStarterUtils; |
| import org.apache.pinot.util.TestUtils; |
| import org.testng.annotations.AfterClass; |
| import org.testng.annotations.BeforeClass; |
| import org.testng.annotations.Test; |
| |
| import static org.testng.Assert.assertEquals; |
| import static org.testng.Assert.assertNotNull; |
| import static org.testng.Assert.assertNull; |
| import static org.testng.Assert.assertTrue; |
| |
| |
| /** |
| * Integration test for minion task of type "MergeRollupTask" |
| */ |
| public class MergeRollupMinionClusterIntegrationTest extends BaseClusterIntegrationTest { |
| private static final String SINGLE_LEVEL_CONCAT_TEST_TABLE = "myTable1"; |
| private static final String SINGLE_LEVEL_ROLLUP_TEST_TABLE = "myTable2"; |
| private static final String MULTI_LEVEL_CONCAT_TEST_TABLE = "myTable3"; |
| private static final String SINGLE_LEVEL_CONCAT_METADATA_TEST_TABLE = "myTable4"; |
| private static final String SINGLE_LEVEL_CONCAT_TEST_REALTIME_TABLE = "myTable5"; |
| private static final String MULTI_LEVEL_CONCAT_PROCESS_ALL_REALTIME_TABLE = "myTable6"; |
| private static final String PROCESS_ALL_MODE_KAFKA_TOPIC = "myKafkaTopic"; |
| private static final long TIMEOUT_IN_MS = 10_000L; |
| |
| protected PinotHelixTaskResourceManager _helixTaskResourceManager; |
| protected PinotTaskManager _taskManager; |
| protected PinotHelixResourceManager _pinotHelixResourceManager; |
| |
| protected final File _segmentDir1 = new File(_tempDir, "segmentDir1"); |
| protected final File _segmentDir2 = new File(_tempDir, "segmentDir2"); |
| protected final File _segmentDir3 = new File(_tempDir, "segmentDir3"); |
| protected final File _segmentDir4 = new File(_tempDir, "segmentDir4"); |
| protected final File _segmentDir5 = new File(_tempDir, "segmentDir5"); |
| protected final File _tarDir1 = new File(_tempDir, "tarDir1"); |
| protected final File _tarDir2 = new File(_tempDir, "tarDir2"); |
| protected final File _tarDir3 = new File(_tempDir, "tarDir3"); |
| protected final File _tarDir4 = new File(_tempDir, "tarDir4"); |
| protected final File _tarDir5 = new File(_tempDir, "tarDir5"); |
| |
| @BeforeClass |
| public void setUp() |
| throws Exception { |
| TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir1, _segmentDir2, _segmentDir3, _segmentDir4, |
| _segmentDir5, _tarDir1, _tarDir2, _tarDir3, _tarDir4, _tarDir5); |
| |
| // Start the Pinot cluster |
| startZk(); |
| startController(); |
| startBrokers(1); |
| startServers(1); |
| // Start Kafka |
| startKafka(); |
| |
| // Create and upload the schema and table config |
| Schema schema = createSchema(); |
| addSchema(schema); |
| schema.setSchemaName(SINGLE_LEVEL_CONCAT_TEST_TABLE); |
| addSchema(schema); |
| schema.setSchemaName(SINGLE_LEVEL_ROLLUP_TEST_TABLE); |
| addSchema(schema); |
| schema.setSchemaName(MULTI_LEVEL_CONCAT_TEST_TABLE); |
| addSchema(schema); |
| schema.setSchemaName(SINGLE_LEVEL_CONCAT_METADATA_TEST_TABLE); |
| addSchema(schema); |
| TableConfig singleLevelConcatTableConfig = |
| createOfflineTableConfig(SINGLE_LEVEL_CONCAT_TEST_TABLE, getSingleLevelConcatTaskConfig()); |
| TableConfig singleLevelRollupTableConfig = |
| createOfflineTableConfig(SINGLE_LEVEL_ROLLUP_TEST_TABLE, getSingleLevelRollupTaskConfig(), |
| getMultiColumnsSegmentPartitionConfig()); |
| TableConfig multiLevelConcatTableConfig = |
| createOfflineTableConfig(MULTI_LEVEL_CONCAT_TEST_TABLE, getMultiLevelConcatTaskConfig()); |
| TableConfig singleLevelConcatMetadataTableConfig = |
| createOfflineTableConfig(SINGLE_LEVEL_CONCAT_METADATA_TEST_TABLE, getSingleLevelConcatMetadataTaskConfig()); |
| addTableConfig(singleLevelConcatTableConfig); |
| addTableConfig(singleLevelRollupTableConfig); |
| addTableConfig(multiLevelConcatTableConfig); |
| addTableConfig(singleLevelConcatMetadataTableConfig); |
| |
| // Unpack the Avro files |
| List<File> avroFiles = unpackAvroData(_tempDir); |
| |
| // Create and upload segments |
| ClusterIntegrationTestUtils.buildSegmentsFromAvro(avroFiles, singleLevelConcatTableConfig, schema, 0, _segmentDir1, |
| _tarDir1); |
| buildSegmentsFromAvroWithPostfix(avroFiles, singleLevelRollupTableConfig, schema, 0, _segmentDir2, _tarDir2, "1"); |
| buildSegmentsFromAvroWithPostfix(avroFiles, singleLevelRollupTableConfig, schema, 0, _segmentDir2, _tarDir2, "2"); |
| ClusterIntegrationTestUtils.buildSegmentsFromAvro(avroFiles, multiLevelConcatTableConfig, schema, 0, _segmentDir3, |
| _tarDir3); |
| ClusterIntegrationTestUtils.buildSegmentsFromAvro(avroFiles, singleLevelConcatMetadataTableConfig, schema, 0, |
| _segmentDir4, _tarDir4); |
| uploadSegments(SINGLE_LEVEL_CONCAT_TEST_TABLE, _tarDir1); |
| uploadSegments(SINGLE_LEVEL_ROLLUP_TEST_TABLE, _tarDir2); |
| uploadSegments(MULTI_LEVEL_CONCAT_TEST_TABLE, _tarDir3); |
| uploadSegments(SINGLE_LEVEL_CONCAT_METADATA_TEST_TABLE, _tarDir4); |
| |
| // create the realtime table |
| TableConfig tableConfig = createRealtimeTableConfig(avroFiles.get(0)); |
| addTableConfig(tableConfig); |
| _kafkaStarters.get(0) |
| .createTopic(PROCESS_ALL_MODE_KAFKA_TOPIC, KafkaStarterUtils.getTopicCreationProps(getNumKafkaPartitions())); |
| schema.setSchemaName(MULTI_LEVEL_CONCAT_PROCESS_ALL_REALTIME_TABLE); |
| addSchema(schema); |
| TableConfig singleLevelConcatProcessAllRealtimeTableConfig = |
| createRealtimeTableConfigWithProcessAllMode(avroFiles.get(0), MULTI_LEVEL_CONCAT_PROCESS_ALL_REALTIME_TABLE, |
| PROCESS_ALL_MODE_KAFKA_TOPIC); |
| addTableConfig(singleLevelConcatProcessAllRealtimeTableConfig); |
| |
| // Push data into Kafka |
| pushAvroIntoKafka(avroFiles); |
| ClusterIntegrationTestUtils.pushAvroIntoKafka(avroFiles.subList(9, 12), "localhost:" + getKafkaPort(), |
| PROCESS_ALL_MODE_KAFKA_TOPIC, getMaxNumKafkaMessagesPerBatch(), getKafkaMessageHeader(), getPartitionColumn(), |
| injectTombstones()); |
| ClusterIntegrationTestUtils.pushAvroIntoKafka(avroFiles.subList(0, 3), "localhost:" + getKafkaPort(), |
| PROCESS_ALL_MODE_KAFKA_TOPIC, getMaxNumKafkaMessagesPerBatch(), getKafkaMessageHeader(), getPartitionColumn(), |
| injectTombstones()); |
| ClusterIntegrationTestUtils.buildSegmentsFromAvro(avroFiles.subList(3, 9), |
| singleLevelConcatProcessAllRealtimeTableConfig, schema, 0, _segmentDir5, _tarDir5); |
| // Wait for all documents loaded |
| waitForAllDocsLoaded(600_000L); |
| |
| // Set up the H2 connection |
| setUpH2Connection(avroFiles); |
| |
| // Initialize the query generator |
| setUpQueryGenerator(avroFiles); |
| |
| startMinion(); |
| _helixTaskResourceManager = _controllerStarter.getHelixTaskResourceManager(); |
| _taskManager = _controllerStarter.getTaskManager(); |
| _pinotHelixResourceManager = _controllerStarter.getHelixResourceManager(); |
| } |
| |
| // this override is used by createRealtimeTableConfig |
| @Override |
| protected String getTableName() { |
| return SINGLE_LEVEL_CONCAT_TEST_REALTIME_TABLE; |
| } |
| |
| // this override is used by createRealtimeTableConfig |
| @Override |
| protected TableTaskConfig getTaskConfig() { |
| Map<String, String> tableTaskConfigs = new HashMap<>(); |
| tableTaskConfigs.put("100days.mergeType", "concat"); |
| tableTaskConfigs.put("100days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("100days.bucketTimePeriod", "100d"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerSegment", "15000"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerTask", "15000"); |
| tableTaskConfigs.put("ActualElapsedTime.aggregationType", "min"); |
| tableTaskConfigs.put("WeatherDelay.aggregationType", "sum"); |
| return new TableTaskConfig(Collections.singletonMap(MinionConstants.MergeRollupTask.TASK_TYPE, tableTaskConfigs)); |
| } |
| |
| private TableConfig createOfflineTableConfig(String tableName, TableTaskConfig taskConfig) { |
| return createOfflineTableConfig(tableName, taskConfig, null); |
| } |
| |
| private TableConfig createOfflineTableConfig(String tableName, TableTaskConfig taskConfig, |
| @Nullable SegmentPartitionConfig partitionConfig) { |
| return new TableConfigBuilder(TableType.OFFLINE).setTableName(tableName).setTimeColumnName(getTimeColumnName()) |
| .setSortedColumn(getSortedColumn()).setInvertedIndexColumns(getInvertedIndexColumns()) |
| .setNoDictionaryColumns(getNoDictionaryColumns()).setRangeIndexColumns(getRangeIndexColumns()) |
| .setBloomFilterColumns(getBloomFilterColumns()).setFieldConfigList(getFieldConfigs()) |
| .setNumReplicas(getNumReplicas()).setSegmentVersion(getSegmentVersion()).setLoadMode(getLoadMode()) |
| .setTaskConfig(taskConfig).setBrokerTenant(getBrokerTenant()).setServerTenant(getServerTenant()) |
| .setIngestionConfig(getIngestionConfig()).setNullHandlingEnabled(getNullHandlingEnabled()) |
| .setSegmentPartitionConfig(partitionConfig).build(); |
| } |
| |
| protected TableConfig createRealtimeTableConfigWithProcessAllMode(File sampleAvroFile, String tableName, |
| String topicName) { |
| AvroFileSchemaKafkaAvroMessageDecoder._avroFile = sampleAvroFile; |
| Map<String, String> streamConfigs = getStreamConfigMap(); |
| streamConfigs.put(StreamConfigProperties.constructStreamProperty("kafka", StreamConfigProperties.STREAM_TOPIC_NAME), |
| topicName); |
| Map<String, String> tableTaskConfigs = new HashMap<>(); |
| tableTaskConfigs.put("100days.mergeType", "concat"); |
| tableTaskConfigs.put("100days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("100days.bucketTimePeriod", "100d"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerSegment", "15000"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerTask", "15000"); |
| tableTaskConfigs.put("200days.mergeType", "concat"); |
| tableTaskConfigs.put("200days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("200days.bucketTimePeriod", "200d"); |
| tableTaskConfigs.put("200days.maxNumRecordsPerSegment", "15000"); |
| tableTaskConfigs.put("200days.maxNumRecordsPerTask", "30000"); |
| tableTaskConfigs.put("ActualElapsedTime.aggregationType", "min"); |
| tableTaskConfigs.put("WeatherDelay.aggregationType", "sum"); |
| tableTaskConfigs.put("mode", "processAll"); |
| return new TableConfigBuilder(TableType.REALTIME).setTableName(tableName).setTimeColumnName(getTimeColumnName()) |
| .setSortedColumn(getSortedColumn()).setInvertedIndexColumns(getInvertedIndexColumns()) |
| .setNoDictionaryColumns(getNoDictionaryColumns()).setRangeIndexColumns(getRangeIndexColumns()) |
| .setBloomFilterColumns(getBloomFilterColumns()).setFieldConfigList(getFieldConfigs()) |
| .setNumReplicas(getNumReplicas()).setSegmentVersion(getSegmentVersion()).setLoadMode(getLoadMode()) |
| .setTaskConfig( |
| new TableTaskConfig(Collections.singletonMap(MinionConstants.MergeRollupTask.TASK_TYPE, tableTaskConfigs))) |
| .setBrokerTenant(getBrokerTenant()).setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig()) |
| .setQueryConfig(getQueryConfig()).setStreamConfigs(streamConfigs) |
| .setNullHandlingEnabled(getNullHandlingEnabled()).build(); |
| } |
| |
| private TableTaskConfig getSingleLevelConcatTaskConfig() { |
| Map<String, String> tableTaskConfigs = new HashMap<>(); |
| tableTaskConfigs.put("100days.mergeType", "concat"); |
| tableTaskConfigs.put("100days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("100days.bucketTimePeriod", "100d"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerSegment", "15000"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerTask", "15000"); |
| tableTaskConfigs.put("ActualElapsedTime.aggregationType", "min"); |
| tableTaskConfigs.put("WeatherDelay.aggregationType", "sum"); |
| tableTaskConfigs.put(BatchConfigProperties.OVERWRITE_OUTPUT, "true"); |
| return new TableTaskConfig(Collections.singletonMap(MinionConstants.MergeRollupTask.TASK_TYPE, tableTaskConfigs)); |
| } |
| |
| private TableTaskConfig getSingleLevelConcatMetadataTaskConfig() { |
| Map<String, String> tableTaskConfigs = new HashMap<>(); |
| tableTaskConfigs.put("100days.mergeType", "concat"); |
| tableTaskConfigs.put("100days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("100days.bucketTimePeriod", "100d"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerSegment", "15000"); |
| tableTaskConfigs.put("100days.maxNumRecordsPerTask", "15000"); |
| tableTaskConfigs.put("ActualElapsedTime.aggregationType", "min"); |
| tableTaskConfigs.put("WeatherDelay.aggregationType", "sum"); |
| tableTaskConfigs.put(BatchConfigProperties.OVERWRITE_OUTPUT, "true"); |
| tableTaskConfigs.put(BatchConfigProperties.PUSH_MODE, BatchConfigProperties.SegmentPushType.METADATA.toString()); |
| return new TableTaskConfig(Collections.singletonMap(MinionConstants.MergeRollupTask.TASK_TYPE, tableTaskConfigs)); |
| } |
| |
| private TableTaskConfig getSingleLevelRollupTaskConfig() { |
| Map<String, String> tableTaskConfigs = new HashMap<>(); |
| tableTaskConfigs.put("150days.mergeType", "rollup"); |
| tableTaskConfigs.put("150days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("150days.bucketTimePeriod", "150d"); |
| tableTaskConfigs.put("150days.roundBucketTimePeriod", "7d"); |
| tableTaskConfigs.put(BatchConfigProperties.OVERWRITE_OUTPUT, "true"); |
| return new TableTaskConfig(Collections.singletonMap(MinionConstants.MergeRollupTask.TASK_TYPE, tableTaskConfigs)); |
| } |
| |
| private TableTaskConfig getMultiLevelConcatTaskConfig() { |
| Map<String, String> tableTaskConfigs = new HashMap<>(); |
| tableTaskConfigs.put("45days.mergeType", "concat"); |
| tableTaskConfigs.put("45days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("45days.bucketTimePeriod", "45d"); |
| tableTaskConfigs.put("45days.maxNumRecordsPerSegment", "100000"); |
| tableTaskConfigs.put("45days.maxNumRecordsPerTask", "100000"); |
| |
| tableTaskConfigs.put("90days.mergeType", "concat"); |
| tableTaskConfigs.put("90days.bufferTimePeriod", "1d"); |
| tableTaskConfigs.put("90days.bucketTimePeriod", "90d"); |
| tableTaskConfigs.put("90days.maxNumRecordsPerSegment", "100000"); |
| tableTaskConfigs.put("90days.maxNumRecordsPerTask", "100000"); |
| tableTaskConfigs.put(BatchConfigProperties.OVERWRITE_OUTPUT, "true"); |
| return new TableTaskConfig(Collections.singletonMap(MinionConstants.MergeRollupTask.TASK_TYPE, tableTaskConfigs)); |
| } |
| |
| private SegmentPartitionConfig getMultiColumnsSegmentPartitionConfig() { |
| Map<String, ColumnPartitionConfig> columnPartitionConfigMap = new HashMap<>(); |
| ColumnPartitionConfig columnOneConfig = new ColumnPartitionConfig("murmur", 1); |
| columnPartitionConfigMap.put("AirlineID", columnOneConfig); |
| ColumnPartitionConfig columnTwoConfig = new ColumnPartitionConfig("murmur", 1); |
| columnPartitionConfigMap.put("Month", columnTwoConfig); |
| return new SegmentPartitionConfig(columnPartitionConfigMap); |
| } |
| |
| private static void buildSegmentsFromAvroWithPostfix(List<File> avroFiles, TableConfig tableConfig, |
| org.apache.pinot.spi.data.Schema schema, int baseSegmentIndex, File segmentDir, File tarDir, String postfix) |
| throws Exception { |
| int numAvroFiles = avroFiles.size(); |
| ExecutorService executorService = Executors.newFixedThreadPool(numAvroFiles); |
| List<Future<Void>> futures = new ArrayList<>(numAvroFiles); |
| for (int i = 0; i < numAvroFiles; i++) { |
| File avroFile = avroFiles.get(i); |
| int segmentIndex = i + baseSegmentIndex; |
| futures.add(executorService.submit(() -> { |
| SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(tableConfig, schema); |
| segmentGeneratorConfig.setInputFilePath(avroFile.getPath()); |
| segmentGeneratorConfig.setOutDir(segmentDir.getPath()); |
| segmentGeneratorConfig.setTableName(tableConfig.getTableName()); |
| // Test segment with space and special character in the file name |
| segmentGeneratorConfig.setSegmentNamePostfix(segmentIndex + "_" + postfix); |
| |
| // Build the segment |
| SegmentIndexCreationDriver driver = new SegmentIndexCreationDriverImpl(); |
| driver.init(segmentGeneratorConfig); |
| driver.build(); |
| |
| // Tar the segment |
| String segmentName = driver.getSegmentName(); |
| File indexDir = new File(segmentDir, segmentName); |
| File segmentTarFile = new File(tarDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION); |
| TarGzCompressionUtils.createTarGzFile(indexDir, segmentTarFile); |
| return null; |
| })); |
| } |
| executorService.shutdown(); |
| for (Future<Void> future : futures) { |
| future.get(); |
| } |
| } |
| |
| /** |
| * Test single level concat task with maxNumRecordPerTask, maxNumRecordPerSegment constraints |
| */ |
| @Test |
| public void testOfflineTableSingleLevelConcat() |
| throws Exception { |
| // The original segments are time partitioned by month: |
| // segmentName (totalDocs) |
| // myTable1_16071_16101_3 (9746) |
| // myTable1_16102_16129_4 (8690) |
| // myTable1_16130_16159_5 (9621) |
| // myTable1_16160_16189_6 (9454) |
| // myTable1_16190_16220_7 (10329) |
| // myTable1_16221_16250_8 (10468) |
| // myTable1_16251_16281_9 (10499) |
| // myTable1_16282_16312_10 (10196) |
| // myTable1_16313_16342_11 (9136) |
| // myTable1_16343_16373_0 (9292) |
| // myTable1_16374_16404_1 (8736) |
| // myTable1_16405_16435_2 (9378) |
| |
| // Expected merge tasks and result segments: |
| // 1. |
| // {myTable1_16071_16101_3} |
| // -> {merged_100days_T1_0_myTable1_16071_16099_0, merged_100days_T1_0_myTable1_16100_16101_1} |
| // 2. |
| // {merged_100days_T1_0_myTable1_16100_16101_1, myTable1_16102_16129_4, myTable1_16130_16159_5} |
| // -> {merged_100days_T2_0_myTable1_16100_???_0(15000), merged_100days_T2_0_myTable1_???_16159_1} |
| // {myTable1_16160_16189_6, myTable1_16190_16220_7} |
| // -> {merged_100days_T2_1_myTable1_16160_16199_0, merged_100days_T2_1_myTable1_16200_16220_1} |
| // 3. |
| // {merged_100days_T2_1_myTable1_16200_16220_1, myTable1_16221_16250_8} |
| // -> {merged_100days_T3_0_myTable1_16200_???_0(15000), merged_100days_T3_0_myTable1_???_16250_1} |
| // {myTable1_16251_16281_9, myTable1_16282_16312_10} |
| // -> {merged_100days_T3_1_myTable1_16251_???_0(15000), merged_100days_T3_1_myTable1_???_16299_1, |
| // merged_100days_T3_1_myTable1_16300_16312_2} |
| // 4. |
| // {merged_100days_T3_1_myTable1_16300_16312_2, myTable1_16313_16342_11, myTable1_16343_16373_0} |
| // -> {merged_100days_T4_0_myTable1_16300_???_0(15000), merged_100days_T4_0_myTable1_???_16373_1} |
| // {myTable1_16374_16404_1} |
| // -> {merged_100days_T4_1_16374_16399_0, merged_100days_T4_1_16400_16404_1} |
| // 5. |
| // {merged_100days_T4_1_16400_16404_1, myTable1_16405_16435_2} |
| // -> {merged_100days_T5_0_myTable1_16400_16435_0} |
| |
| String sqlQuery = "SELECT count(*) FROM myTable1"; // 115545 rows for the test table |
| JsonNode expectedJson = postQuery(sqlQuery); |
| int[] expectedNumSubTasks = {1, 2, 2, 2, 1}; |
| int[] expectedNumSegmentsQueried = {13, 12, 13, 13, 12}; |
| long expectedWatermark = 16000 * 86_400_000L; |
| String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(SINGLE_LEVEL_CONCAT_TEST_TABLE); |
| int numTasks = 0; |
| List<String> taskList; |
| for (String tasks = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .get(0); tasks != null; taskList = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE), |
| tasks = taskList != null ? taskList.get(0) : null, numTasks++) { |
| assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); |
| assertTrue(_helixTaskResourceManager.getTaskQueues() |
| .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); |
| // Will not schedule task if there's incomplete task |
| assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) |
| .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); |
| waitForTaskToComplete(); |
| |
| // Check watermark |
| MergeRollupTaskMetadata minionTaskMetadata = MergeRollupTaskMetadata.fromZNRecord( |
| _taskManager.getClusterInfoAccessor() |
| .getMinionTaskMetadataZNRecord(MinionConstants.MergeRollupTask.TASK_TYPE, offlineTableName)); |
| assertNotNull(minionTaskMetadata); |
| assertEquals((long) minionTaskMetadata.getWatermarkMap().get("100days"), expectedWatermark); |
| expectedWatermark += 100 * 86_400_000L; |
| |
| // Check metadata of merged segments |
| for (SegmentZKMetadata metadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) { |
| if (metadata.getSegmentName().startsWith("merged")) { |
| // Check merged segment zk metadata |
| assertNotNull(metadata.getCustomMap()); |
| assertEquals("100days", |
| metadata.getCustomMap().get(MinionConstants.MergeRollupTask.SEGMENT_ZK_METADATA_MERGE_LEVEL_KEY)); |
| // Check merged segments are time partitioned |
| assertEquals(metadata.getEndTimeMs() / (86_400_000L * 100), metadata.getStartTimeMs() / (86_400_000L * 100)); |
| } |
| } |
| |
| final int finalNumTasks = numTasks; |
| TestUtils.waitForCondition(aVoid -> { |
| try { |
| // Check num total doc of merged segments are the same as the original segments |
| JsonNode actualJson = postQuery(sqlQuery); |
| if (!SqlResultComparator.areEqual(actualJson, expectedJson, sqlQuery)) { |
| return false; |
| } |
| // Check query routing |
| int numSegmentsQueried = actualJson.get("numSegmentsQueried").asInt(); |
| return numSegmentsQueried == expectedNumSegmentsQueried[finalNumTasks]; |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| }, TIMEOUT_IN_MS, "Timeout while validating segments"); |
| } |
| // Check total tasks |
| assertEquals(numTasks, 5); |
| |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskDelayInNumBuckets.myTable1_OFFLINE.100days")); |
| // Drop the table |
| dropOfflineTable(SINGLE_LEVEL_CONCAT_TEST_TABLE); |
| |
| // Check if the task metadata is cleaned up on table deletion |
| verifyTableDelete(offlineTableName); |
| } |
| |
| /** |
| * Test single level concat task with maxNumRecordPerTask, maxNumRecordPerSegment constraints |
| * Push type is set to Metadata |
| */ |
| @Test |
| public void testOfflineTableSingleLevelConcatWithMetadataPush() |
| throws Exception { |
| // The original segments are time partitioned by month: |
| // segmentName (totalDocs) |
| // myTable1_16071_16101_3 (9746) |
| // myTable1_16102_16129_4 (8690) |
| // myTable1_16130_16159_5 (9621) |
| // myTable1_16160_16189_6 (9454) |
| // myTable1_16190_16220_7 (10329) |
| // myTable1_16221_16250_8 (10468) |
| // myTable1_16251_16281_9 (10499) |
| // myTable1_16282_16312_10 (10196) |
| // myTable1_16313_16342_11 (9136) |
| // myTable1_16343_16373_0 (9292) |
| // myTable1_16374_16404_1 (8736) |
| // myTable1_16405_16435_2 (9378) |
| |
| // Expected merge tasks and result segments: |
| // 1. |
| // {myTable1_16071_16101_3} |
| // -> {merged_100days_T1_0_myTable1_16071_16099_0, merged_100days_T1_0_myTable1_16100_16101_1} |
| // 2. |
| // {merged_100days_T1_0_myTable1_16100_16101_1, myTable1_16102_16129_4, myTable1_16130_16159_5} |
| // -> {merged_100days_T2_0_myTable1_16100_???_0(15000), merged_100days_T2_0_myTable1_???_16159_1} |
| // {myTable1_16160_16189_6, myTable1_16190_16220_7} |
| // -> {merged_100days_T2_1_myTable1_16160_16199_0, merged_100days_T2_1_myTable1_16200_16220_1} |
| // 3. |
| // {merged_100days_T2_1_myTable1_16200_16220_1, myTable1_16221_16250_8} |
| // -> {merged_100days_T3_0_myTable1_16200_???_0(15000), merged_100days_T3_0_myTable1_???_16250_1} |
| // {myTable1_16251_16281_9, myTable1_16282_16312_10} |
| // -> {merged_100days_T3_1_myTable1_16251_???_0(15000), merged_100days_T3_1_myTable1_???_16299_1, |
| // merged_100days_T3_1_myTable1_16300_16312_2} |
| // 4. |
| // {merged_100days_T3_1_myTable1_16300_16312_2, myTable1_16313_16342_11, myTable1_16343_16373_0} |
| // -> {merged_100days_T4_0_myTable1_16300_???_0(15000), merged_100days_T4_0_myTable1_???_16373_1} |
| // {myTable1_16374_16404_1} |
| // -> {merged_100days_T4_1_16374_16399_0, merged_100days_T4_1_16400_16404_1} |
| // 5. |
| // {merged_100days_T4_1_16400_16404_1, myTable1_16405_16435_2} |
| // -> {merged_100days_T5_0_myTable1_16400_16435_0} |
| |
| String sqlQuery = "SELECT count(*) FROM myTable4"; // 115545 rows for the test table |
| JsonNode expectedJson = postQuery(sqlQuery); |
| int[] expectedNumSubTasks = {1, 2, 2, 2, 1}; |
| int[] expectedNumSegmentsQueried = {13, 12, 13, 13, 12}; |
| long expectedWatermark = 16000 * 86_400_000L; |
| String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(SINGLE_LEVEL_CONCAT_METADATA_TEST_TABLE); |
| int numTasks = 0; |
| List<String> taskList; |
| for (String tasks = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .get(0); tasks != null; taskList = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE), |
| tasks = taskList != null ? taskList.get(0) : null, numTasks++) { |
| assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); |
| assertTrue(_helixTaskResourceManager.getTaskQueues() |
| .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); |
| // Will not schedule task if there's incomplete task |
| assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) |
| .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); |
| waitForTaskToComplete(); |
| |
| // Check watermark |
| MergeRollupTaskMetadata minionTaskMetadata = MergeRollupTaskMetadata.fromZNRecord( |
| _taskManager.getClusterInfoAccessor() |
| .getMinionTaskMetadataZNRecord(MinionConstants.MergeRollupTask.TASK_TYPE, offlineTableName)); |
| assertNotNull(minionTaskMetadata); |
| assertEquals((long) minionTaskMetadata.getWatermarkMap().get("100days"), expectedWatermark); |
| expectedWatermark += 100 * 86_400_000L; |
| |
| // Check metadata of merged segments |
| for (SegmentZKMetadata metadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) { |
| if (metadata.getSegmentName().startsWith("merged")) { |
| // Check merged segment zk metadata |
| assertNotNull(metadata.getCustomMap()); |
| assertEquals("100days", |
| metadata.getCustomMap().get(MinionConstants.MergeRollupTask.SEGMENT_ZK_METADATA_MERGE_LEVEL_KEY)); |
| // Check merged segments are time partitioned |
| assertEquals(metadata.getEndTimeMs() / (86_400_000L * 100), metadata.getStartTimeMs() / (86_400_000L * 100)); |
| } |
| } |
| |
| final int finalNumTasks = numTasks; |
| TestUtils.waitForCondition(aVoid -> { |
| try { |
| // Check num total doc of merged segments are the same as the original segments |
| JsonNode actualJson = postQuery(sqlQuery); |
| if (!SqlResultComparator.areEqual(actualJson, expectedJson, sqlQuery)) { |
| return false; |
| } |
| // Check query routing |
| int numSegmentsQueried = actualJson.get("numSegmentsQueried").asInt(); |
| return numSegmentsQueried == expectedNumSegmentsQueried[finalNumTasks]; |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| }, TIMEOUT_IN_MS, "Timeout while validating segments"); |
| } |
| // Check total tasks |
| assertEquals(numTasks, 5); |
| |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskDelayInNumBuckets.myTable4_OFFLINE.100days")); |
| |
| // Drop the table |
| dropOfflineTable(SINGLE_LEVEL_CONCAT_METADATA_TEST_TABLE); |
| |
| // Check if the task metadata is cleaned up on table deletion |
| verifyTableDelete(offlineTableName); |
| } |
| |
| /** |
| * Test single level rollup task with duplicate data (original segments * 2) |
| */ |
| @Test |
| public void testOfflineTableSingleLevelRollup() |
| throws Exception { |
| // The original segments are time partitioned by month: |
| // segmentName (totalDocs) |
| // myTable2_16071_16101_3_1, myTable2_16071_16101_3_2 (9746) |
| // myTable2_16102_16129_4_1, myTable2_16102_16129_4_2 (8690) |
| // myTable2_16130_16159_5_1, myTable2_16130_16159_5_2 (9621) |
| // myTable2_16160_16189_6_1, myTable2_16160_16189_6_2 (9454) |
| // myTable2_16190_16220_7_1, myTable2_16190_16220_7_2 (10329) |
| // myTable2_16221_16250_8_1, myTable2_16221_16250_8_2 (10468) |
| // myTable2_16251_16281_9_1, myTable2_16251_16281_9_2 (10499) |
| // myTable2_16282_16312_10_1, myTable2_16282_16312_10_2 (10196) |
| // myTable2_16313_16342_11_1, myTable2_16313_16342_11_2 (9136) |
| // myTable2_16343_16373_0_1, myTable2_16343_16373_0_2 (9292) |
| // myTable2_16374_16404_1_1, myTable2_16374_16404_1_2 (8736) |
| // myTable2_16405_16435_2_1, myTable2_16405_16435_2_2 (9378) |
| |
| // Expected merge tasks and result segments: |
| // 1. |
| // {myTable2_16071_16101_3_1, myTable2_16071_16101_3_2, myTable2_16102_16129_4_1, myTable2_16102_16129_4_2, |
| // myTable2_16130_16159_5_1, myTable2_16130_16159_5_2, myTable2_16160_16189_6_1, myTable2_16160_16189_6_2 |
| // myTable2_16190_16220_7} |
| // -> {merged_150days_T1_0_myTable2_16065_16198_0, merged_150days_T1_0_myTable2_16205_16219_1} |
| // 2. |
| // {merged_150days_T1_0_myTable2_16205_16219_1, myTable2_16221_16250_8_1, myTable2_16221_16250_8_2, |
| // myTable2_16251_16281_9_1, myTable2_16251_16281_9_2, myTable2_16282_16312_10_1 |
| // myTable2_16282_16312_10_2, myTable2_16313_16342_11_1, myTable2_16313_16342_11_2, |
| // myTable2_16343_16373_0_1, myTable2_16343_16373_0_2} |
| // -> {merged_150days_1628644088146_0_myTable2_16205_16345_0, |
| // merged_150days_1628644088146_0_myTable2_16352_16373_1} |
| // 3. |
| // {merged_150days_1628644088146_0_myTable2_16352_16373_1, myTable2_16374_16404_1_1, myTable2_16374_16404_1_2 |
| // myTable2_16405_16435_2_1, myTable2_16405_16435_2_2} |
| // -> {merged_150days_1628644105127_0_myTable2_16352_16429_0} |
| |
| String sqlQuery = "SELECT count(*) FROM myTable2"; // 115545 rows for the test table |
| JsonNode expectedJson = postQuery(sqlQuery); |
| int[] expectedNumSegmentsQueried = {16, 7, 3}; |
| long expectedWatermark = 16050 * 86_400_000L; |
| String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(SINGLE_LEVEL_ROLLUP_TEST_TABLE); |
| int numTasks = 0; |
| List<String> taskList; |
| for (String tasks = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .get(0); tasks != null; taskList = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE), |
| tasks = taskList != null ? taskList.get(0) : null, numTasks++) { |
| assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), 1); |
| assertTrue(_helixTaskResourceManager.getTaskQueues() |
| .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); |
| // Will not schedule task if there's incomplete task |
| assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) |
| .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); |
| waitForTaskToComplete(); |
| |
| // Check watermark |
| MergeRollupTaskMetadata minionTaskMetadata = MergeRollupTaskMetadata.fromZNRecord( |
| _taskManager.getClusterInfoAccessor() |
| .getMinionTaskMetadataZNRecord(MinionConstants.MergeRollupTask.TASK_TYPE, offlineTableName)); |
| assertNotNull(minionTaskMetadata); |
| assertEquals((long) minionTaskMetadata.getWatermarkMap().get("150days"), expectedWatermark); |
| expectedWatermark += 150 * 86_400_000L; |
| |
| // Check metadata of merged segments |
| for (SegmentZKMetadata metadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) { |
| if (metadata.getSegmentName().startsWith("merged")) { |
| // Check merged segment zk metadata |
| assertNotNull(metadata.getCustomMap()); |
| assertEquals("150days", |
| metadata.getCustomMap().get(MinionConstants.MergeRollupTask.SEGMENT_ZK_METADATA_MERGE_LEVEL_KEY)); |
| // Check merged segments are time partitioned |
| assertEquals(metadata.getEndTimeMs() / (86_400_000L * 150), metadata.getStartTimeMs() / (86_400_000L * 150)); |
| } |
| } |
| |
| final int finalNumTasks = numTasks; |
| TestUtils.waitForCondition(aVoid -> { |
| try { |
| // Check total doc of merged segments are less than the original segments |
| JsonNode actualJson = postQuery(sqlQuery); |
| if (actualJson.get("resultTable").get("rows").get(0).get(0).asInt() >= expectedJson.get("resultTable") |
| .get("rows").get(0).get(0).asInt()) { |
| return false; |
| } |
| // Check query routing |
| int numSegmentsQueried = actualJson.get("numSegmentsQueried").asInt(); |
| return numSegmentsQueried == expectedNumSegmentsQueried[finalNumTasks]; |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| }, TIMEOUT_IN_MS, "Timeout while validating segments"); |
| } |
| |
| // Check total doc is half of the original after all merge tasks are finished |
| JsonNode actualJson = postQuery(sqlQuery); |
| assertEquals(actualJson.get("resultTable").get("rows").get(0).get(0).asInt(), |
| expectedJson.get("resultTable").get("rows").get(0).get(0).asInt() / 2); |
| // Check time column is rounded |
| JsonNode responseJson = |
| postQuery("SELECT count(*), DaysSinceEpoch FROM myTable2 GROUP BY DaysSinceEpoch ORDER BY DaysSinceEpoch"); |
| for (int i = 0; i < responseJson.get("resultTable").get("rows").size(); i++) { |
| int daysSinceEpoch = responseJson.get("resultTable").get("rows").get(i).get(1).asInt(); |
| assertTrue(daysSinceEpoch % 7 == 0); |
| } |
| // Check total tasks |
| assertEquals(numTasks, 3); |
| |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskDelayInNumBuckets.myTable2_OFFLINE.150days")); |
| } |
| |
| /** |
| * Test multi level concat task |
| */ |
| @Test |
| public void testOfflineTableMultiLevelConcat() |
| throws Exception { |
| // The original segments are time partitioned by month: |
| // segmentName (totalDocs) |
| // myTable3_16071_16101_3 (9746) |
| // myTable3_16102_16129_4 (8690) |
| // myTable3_16130_16159_5 (9621) |
| // myTable3_16160_16189_6 (9454) |
| // myTable3_16190_16220_7 (10329) |
| // myTable3_16221_16250_8 (10468) |
| // myTable3_16251_16281_9 (10499) |
| // myTable3_16282_16312_10 (10196) |
| // myTable3_16313_16342_11 (9136) |
| // myTable3_16343_16373_0 (9292) |
| // myTable3_16374_16404_1 (8736) |
| // myTable3_16405_16435_2 (9378) |
| |
| // Expected merge tasks and results: |
| // 1. |
| // 45days: {myTable3_16071_16101_3, myTable3_16102_16129_4} |
| // -> {merged_45days_T1_0_myTable3_16071_16109_0, merged_45days_T1_0_myTable3_16110_16129_1} |
| // watermark: {45days: 16065, 90days: null} |
| // 2. |
| // 45days: {merged_45days_T1_0_myTable3_16110_16129_1, myTable3_16130_16159_5} |
| // -> {merged_45days_T2_0_myTable3_16110_16154_0, merged_45days_T2_0_myTable3_16155_16159_1} |
| // 90days: {merged_45days_T1_0_myTable3_16071_16109_0} |
| // -> {merged_90days_T2_0_myTable3_16071_16109_0} |
| // watermark: {45days: 16110, 90days: 16020} |
| // 3. |
| // 45days: {merged_45days_T2_0_myTable3_16155_16159_1, myTable3_16160_16189_6, myTable3_16190_16220_7} |
| // -> {merged_45days_T3_0_myTable3_16155_16199_0, merged_45days_T3_0_myTable3_16200_16220_1} |
| // watermark: {45days: 16155, 90days: 16020} |
| // 4. |
| // 45days: {merged_45days_T3_-_myTable3_16200_16220_1, myTable3_16221_16250_8} |
| // -> {merged_45days_T4_0_myTable3_16200_16244_0, merged_45days_T4_0_myTable3_16245_16250_1} |
| // 90days: {merged_45days_T2_0_myTable3_16110_16154_0, merged_45days_T3_0_myTable3_16155_16199_0} |
| // -> {merged_90days_T4_0_myTable3_16110_16199_0} |
| // watermark: {45days: 16200, 90days: 16110} |
| // 5. |
| // 45days: {merged_45days_T4_0_myTable3_16245_16250_1, myTable3_16251_16281_9, myTable3_16282_16312_10} |
| // -> {merged_45days_T5_0_myTable3_16245_16289_0, merged_45days_T5_0_myTable3_16290_16312_1} |
| // watermark: {45days: 16245, 90days: 16110} |
| // 6. |
| // 45days: {merged_45days_T5_0_myTable3_16290_16312_1, myTable3_16313_16342_11} |
| // -> {merged_45days_T6_0_myTable3_16290_16334_0, merged_45days_T6_0_myTable3_16335_16342_1} |
| // 90days: {merged_45days_T4_0_myTable3_16200_16244_0, merged_45days_T5_0_myTable3_16245_16289_0} |
| // -> {merged_90days_T6_0_myTable3_16200_16289_0} |
| // watermark: {45days: 16290, 90days: 16200} |
| // 7. |
| // 45days: {merged_45days_T6_0_myTable3_16335_16342_1, myTable_16343_16373_0, myTable_16374_16404_1} |
| // -> {merged_45days_T7_0_myTable3_16335_16379_0, merged_45days_T7_0_myTable3_16380_16404_1} |
| // watermark: {45days: 16335, 90days: 16200} |
| // 8. |
| // 45days: {merged_45days_T7_0_myTable3_16380_16404_1, myTable3_16405_16435_2} |
| // -> {merged_45days_T8_0_myTable3_16380_16424_0, merged_45days_T8_1_myTable3_16425_16435_1} |
| // 90days: {merged_45days_T6_0_myTable3_16290_16334_0, merged_45days_T7_0_myTable3_16335_16379_0} |
| // -> {merged_90days_T8_0_myTable3_16290_16379_0} |
| // watermark: {45days:16380, 90days: 16290} |
| // 9. |
| // 45days: no segment left, not scheduling |
| // 90days: [16380, 16470) is not a valid merge window because windowEndTime > 45days watermark, not scheduling |
| |
| String sqlQuery = "SELECT count(*) FROM myTable3"; // 115545 rows for the test table |
| JsonNode expectedJson = postQuery(sqlQuery); |
| int[] expectedNumSubTasks = {1, 2, 1, 2, 1, 2, 1, 2, 1}; |
| int[] expectedNumSegmentsQueried = {12, 12, 11, 10, 9, 8, 7, 6, 5}; |
| Long[] expectedWatermarks45Days = {16065L, 16110L, 16155L, 16200L, 16245L, 16290L, 16335L, 16380L}; |
| Long[] expectedWatermarks90Days = {null, 16020L, 16020L, 16110L, 16110L, 16200L, 16200L, 16290L}; |
| for (int i = 0; i < expectedWatermarks45Days.length; i++) { |
| expectedWatermarks45Days[i] *= 86_400_000L; |
| } |
| for (int i = 1; i < expectedWatermarks90Days.length; i++) { |
| expectedWatermarks90Days[i] *= 86_400_000L; |
| } |
| |
| String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(MULTI_LEVEL_CONCAT_TEST_TABLE); |
| int numTasks = 0; |
| List<String> taskList; |
| for (String tasks = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .get(0); tasks != null; taskList = |
| _taskManager.scheduleAllTasksForTable(offlineTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE), |
| tasks = taskList != null ? taskList.get(0) : null, numTasks++) { |
| assertEquals(_helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); |
| assertTrue(_helixTaskResourceManager.getTaskQueues() |
| .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); |
| // Will not schedule task if there's incomplete task |
| assertNull(_taskManager.scheduleAllTasksForTable(offlineTableName, null) |
| .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); |
| waitForTaskToComplete(); |
| |
| // Check watermark |
| MergeRollupTaskMetadata minionTaskMetadata = MergeRollupTaskMetadata.fromZNRecord( |
| _taskManager.getClusterInfoAccessor() |
| .getMinionTaskMetadataZNRecord(MinionConstants.MergeRollupTask.TASK_TYPE, offlineTableName)); |
| assertNotNull(minionTaskMetadata); |
| assertEquals(minionTaskMetadata.getWatermarkMap().get("45days"), expectedWatermarks45Days[numTasks]); |
| assertEquals(minionTaskMetadata.getWatermarkMap().get("90days"), expectedWatermarks90Days[numTasks]); |
| |
| // Check metadata of merged segments |
| for (SegmentZKMetadata metadata : _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName)) { |
| if (metadata.getSegmentName().startsWith("merged")) { |
| // Check merged segment zk metadata |
| assertNotNull(metadata.getCustomMap()); |
| if (metadata.getSegmentName().startsWith("merged_45days")) { |
| assertEquals("45days", |
| metadata.getCustomMap().get(MinionConstants.MergeRollupTask.SEGMENT_ZK_METADATA_MERGE_LEVEL_KEY)); |
| assertEquals(metadata.getEndTimeMs() / (86_400_000L * 45), metadata.getStartTimeMs() / (86_400_000L * 45)); |
| } |
| if (metadata.getSegmentName().startsWith("merged_90days")) { |
| assertEquals("90days", |
| metadata.getCustomMap().get(MinionConstants.MergeRollupTask.SEGMENT_ZK_METADATA_MERGE_LEVEL_KEY)); |
| assertEquals(metadata.getEndTimeMs() / (86_400_000L * 90), metadata.getStartTimeMs() / (86_400_000L * 90)); |
| } |
| } |
| } |
| |
| final int finalNumTasks = numTasks; |
| TestUtils.waitForCondition(aVoid -> { |
| try { |
| // Check total doc of merged segments are the same as the original segments |
| JsonNode actualJson = postQuery(sqlQuery); |
| if (!SqlResultComparator.areEqual(actualJson, expectedJson, sqlQuery)) { |
| return false; |
| } |
| // Check query routing |
| int numSegmentsQueried = actualJson.get("numSegmentsQueried").asInt(); |
| return numSegmentsQueried == expectedNumSegmentsQueried[finalNumTasks]; |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| }, TIMEOUT_IN_MS, "Timeout while validating segments"); |
| } |
| // Check total tasks |
| assertEquals(numTasks, 8); |
| |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskDelayInNumBuckets.myTable3_OFFLINE.45days")); |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskDelayInNumBuckets.myTable3_OFFLINE.90days")); |
| } |
| |
| protected void verifyTableDelete(String tableNameWithType) { |
| TestUtils.waitForCondition(input -> { |
| // Check if the segment lineage is cleaned up |
| if (SegmentLineageAccessHelper.getSegmentLineage(_propertyStore, tableNameWithType) != null) { |
| return false; |
| } |
| // Check if the task metadata is cleaned up |
| if (MinionTaskMetadataUtils.fetchTaskMetadata(_propertyStore, MinionConstants.MergeRollupTask.TASK_TYPE, |
| tableNameWithType) != null) { |
| return false; |
| } |
| return true; |
| }, 1_000L, 60_000L, "Failed to delete table"); |
| } |
| |
| protected void waitForTaskToComplete() { |
| TestUtils.waitForCondition(input -> { |
| // Check task state |
| for (TaskState taskState : _helixTaskResourceManager.getTaskStates(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .values()) { |
| if (taskState != TaskState.COMPLETED) { |
| return false; |
| } |
| } |
| return true; |
| }, 600_000L, "Failed to complete task"); |
| } |
| |
| // The use case is similar as the one defined in offline table |
| @Test |
| public void testRealtimeTableSingleLevelConcat() |
| throws Exception { |
| // The original segments: |
| // mytable__0__0__{ts00} ... mytable__0__23__{ts023} |
| // mytable__1__0__{ts10} ... mytable__1__22__{ts122} |
| // |
| // Expected result segments: |
| // merged_100days_{ts1}_0_mytable_16071_16099_0 |
| // merged_100days_{ts2}_0_mytable_16100_16154_0 |
| // merged_100days_{ts2}_0_mytable_16101_16146_1 |
| // merged_100days_{ts2}_1_mytable_16147_16199_0 |
| // merged_100days_{ts2}_2_mytable_16196_16199_0 |
| // merged_100days_{ts3}_0_mytable_16200_16252_1 |
| // merged_100days_{ts3}_0_mytable_16200_16252_0 |
| // merged_100days_{ts3}_1_mytable_16245_16295_0 |
| // merged_100days_{ts3}_2_mytable_16290_16299_0 |
| // merged_100days_{ts4}_0_mytable_16300_16359_0 |
| // merged_100days_{ts4}_0_mytable_16323_16345_1 |
| // merged_100days_{ts4}_1_mytable_16358_16399_0 |
| // merged_100days_{ts5}_0_mytable_16400_16435_0 |
| // mytable__0__23__{ts023} (in progress) |
| // mytable__1__22__{ts122} (in progress) |
| PinotHelixTaskResourceManager helixTaskResourceManager = _controllerStarter.getHelixTaskResourceManager(); |
| PinotTaskManager taskManager = _controllerStarter.getTaskManager(); |
| PinotHelixResourceManager pinotHelixResourceManager = _controllerStarter.getHelixResourceManager(); |
| String tableName = getTableName(); |
| |
| String sqlQuery = "SELECT count(*) FROM " + tableName; // 115545 rows for the test table |
| JsonNode expectedJson = postQuery(sqlQuery); |
| // disable some checks for now because github does not generate the same number of tasks sometimes |
| // need to figure out why they work locally all the time but not on github |
| // I feel it maybe related to resources or timestamps |
| // int[] expectedNumSubTasks = {1, 3, 3, 2, 1}; |
| // int[] expectedNumSegmentsQueried = {44, 37, 26, 18, 15}; |
| long expectedWatermark = 16000 * 86_400_000L; |
| String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(tableName); |
| int numTasks = 0; |
| List<String> taskList; |
| for (String tasks = |
| taskManager.scheduleAllTasksForTable(realtimeTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .get(0); tasks != null; taskList = |
| taskManager.scheduleAllTasksForTable(realtimeTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE), |
| tasks = taskList != null ? taskList.get(0) : null, numTasks++) { |
| // assertEquals(helixTaskResourceManager.getSubtaskConfigs(tasks).size(), expectedNumSubTasks[numTasks]); |
| assertTrue(helixTaskResourceManager.getTaskQueues() |
| .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); |
| |
| // Will not schedule task if there's incomplete task |
| assertNull(taskManager.scheduleAllTasksForTable(realtimeTableName, null) |
| .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); |
| waitForTaskToComplete(); |
| |
| // Check watermark |
| MergeRollupTaskMetadata minionTaskMetadata = MergeRollupTaskMetadata.fromZNRecord( |
| taskManager.getClusterInfoAccessor() |
| .getMinionTaskMetadataZNRecord(MinionConstants.MergeRollupTask.TASK_TYPE, realtimeTableName)); |
| assertNotNull(minionTaskMetadata); |
| assertEquals((long) minionTaskMetadata.getWatermarkMap().get("100days"), expectedWatermark); |
| expectedWatermark += 100 * 86_400_000L; |
| |
| // Check metadata of merged segments |
| for (SegmentZKMetadata metadata : pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName)) { |
| if (metadata.getSegmentName().startsWith("merged")) { |
| // Check merged segment zk metadata |
| assertNotNull(metadata.getCustomMap()); |
| assertEquals("100days", |
| metadata.getCustomMap().get(MinionConstants.MergeRollupTask.SEGMENT_ZK_METADATA_MERGE_LEVEL_KEY)); |
| // Check merged segments are time partitioned |
| assertEquals(metadata.getEndTimeMs() / (86_400_000L * 100), metadata.getStartTimeMs() / (86_400_000L * 100)); |
| } |
| } |
| |
| final int finalNumTasks = numTasks; |
| TestUtils.waitForCondition(aVoid -> { |
| try { |
| // Check num total doc of merged segments are the same as the original segments |
| JsonNode actualJson = postQuery(sqlQuery); |
| if (!SqlResultComparator.areEqual(actualJson, expectedJson, sqlQuery)) { |
| return false; |
| } |
| // Check query routing |
| // int numSegmentsQueried = actualJson.get("numSegmentsQueried").asInt(); |
| // return numSegmentsQueried == expectedNumSegmentsQueried[finalNumTasks] |
| // // when running on github tests, the consumer sometimes queries one more segment |
| // || numSegmentsQueried == expectedNumSegmentsQueried[finalNumTasks] + 1; |
| return true; |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| }, TIMEOUT_IN_MS, "Timeout while validating segments"); |
| } |
| // Check total tasks |
| assertEquals(numTasks, 5); |
| |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskDelayInNumBuckets.myTable5_REALTIME.100days")); |
| |
| // Drop the table |
| dropRealtimeTable(tableName); |
| |
| // Check if the task metadata is cleaned up on table deletion |
| verifyTableDelete(realtimeTableName); |
| } |
| |
| @Test |
| public void testRealtimeTableProcessAllModeMultiLevelConcat() |
| throws Exception { |
| // The original segments (time range: [16251, 16428]): |
| // mytable__0__0__{ts00} ... mytable__0__11__{ts011} |
| // mytable__1__0__{ts00} ... mytable__1__11__{ts111} |
| // |
| // Scheduled time ranges and number buckets to process: |
| // round | 100days 200days |
| // 1 | [16251, 16299], 3 | [] , 0 |
| // 2 | [16300, 16399], 2 | [] , 0 |
| // 3 | [16400, 16428], 1 | [16251, 16399], 1 |
| // 4 | [] , 0 | [16400, 16428], 1 |
| // |
| // Upload historical segments manually: |
| // myTable_16071_16101_0 |
| // myTable_16102_16129_1 |
| // myTable_16130_16159_2 |
| // myTable_16160_16189_3 |
| // myTable_16190_16220_4 |
| // myTable_16221_16250_5 |
| // |
| // Scheduled time ranges and number buckets to process: |
| // round | 100days 200days |
| // 5 | [16071, 16099], 3 | [] , 0 |
| // 6 | [16100, 16199], 2 | [] , 0 |
| // 7 | [16200, 16299], 1 | [16071, 16199], 1 |
| // 8 | [] , 0 | [16200, 16399], 1 |
| PinotHelixTaskResourceManager helixTaskResourceManager = _controllerStarter.getHelixTaskResourceManager(); |
| PinotTaskManager taskManager = _controllerStarter.getTaskManager(); |
| String tableName = MULTI_LEVEL_CONCAT_PROCESS_ALL_REALTIME_TABLE; |
| |
| String sqlQuery = "SELECT count(*) FROM " + tableName; |
| JsonNode expectedJson = postQuery(sqlQuery); |
| long[] expectedNumBucketsToProcess100Days = {3, 2, 1, 0, 3, 2, 1, 0}; |
| long[] expectedNumBucketsToProcess200Days = {0, 0, 1, 1, 0, 0, 1, 1}; |
| String realtimeTableName = TableNameBuilder.REALTIME.tableNameWithType(tableName); |
| int numTasks = 0; |
| List<String> taskList; |
| for (String tasks = |
| taskManager.scheduleAllTasksForTable(realtimeTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .get(0); tasks != null; taskList = |
| taskManager.scheduleAllTasksForTable(realtimeTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE), |
| tasks = taskList != null ? taskList.get(0) : null, numTasks++) { |
| assertTrue(helixTaskResourceManager.getTaskQueues() |
| .contains(PinotHelixTaskResourceManager.getHelixJobQueueName(MinionConstants.MergeRollupTask.TASK_TYPE))); |
| |
| // Will not schedule task if there's incomplete task |
| assertNull(taskManager.scheduleAllTasksForTable(realtimeTableName, null) |
| .get(MinionConstants.RealtimeToOfflineSegmentsTask.TASK_TYPE)); |
| waitForTaskToComplete(); |
| |
| // Check not using watermarks |
| ZNRecord minionTaskMetadataZNRecord = taskManager.getClusterInfoAccessor() |
| .getMinionTaskMetadataZNRecord(MinionConstants.MergeRollupTask.TASK_TYPE, realtimeTableName); |
| assertNull(minionTaskMetadataZNRecord); |
| |
| // Check metrics |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskNumBucketsToProcess.myTable6_REALTIME.100days")); |
| assertTrue(MetricValueUtils.gaugeExists(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskNumBucketsToProcess.myTable6_REALTIME.200days")); |
| long numBucketsToProcess = MetricValueUtils.getGaugeValue(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskNumBucketsToProcess.myTable6_REALTIME.100days"); |
| assertEquals(numBucketsToProcess, expectedNumBucketsToProcess100Days[numTasks]); |
| numBucketsToProcess = MetricValueUtils.getGaugeValue(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskNumBucketsToProcess.myTable6_REALTIME.200days"); |
| assertEquals(numBucketsToProcess, expectedNumBucketsToProcess200Days[numTasks]); |
| } |
| // Check total tasks |
| assertEquals(numTasks, 4); |
| |
| // Check query results |
| JsonNode actualJson = postQuery(sqlQuery); |
| assertTrue(SqlResultComparator.areEqual(actualJson, expectedJson, sqlQuery)); |
| |
| // Upload historical data and schedule |
| uploadSegments(MULTI_LEVEL_CONCAT_PROCESS_ALL_REALTIME_TABLE, TableType.REALTIME, _tarDir5); |
| waitForAllDocsLoaded(600_000L); |
| |
| for (String tasks = |
| taskManager.scheduleAllTasksForTable(realtimeTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE) |
| .get(0); tasks != null; taskList = |
| taskManager.scheduleAllTasksForTable(realtimeTableName, null).get(MinionConstants.MergeRollupTask.TASK_TYPE), |
| tasks = taskList != null ? taskList.get(0) : null, numTasks++) { |
| waitForTaskToComplete(); |
| // Check metrics |
| long numBucketsToProcess = MetricValueUtils.getGaugeValue(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskNumBucketsToProcess.myTable6_REALTIME.100days"); |
| assertEquals(numBucketsToProcess, expectedNumBucketsToProcess100Days[numTasks]); |
| numBucketsToProcess = MetricValueUtils.getGaugeValue(_controllerStarter.getControllerMetrics(), |
| "mergeRollupTaskNumBucketsToProcess.myTable6_REALTIME.200days"); |
| assertEquals(numBucketsToProcess, expectedNumBucketsToProcess200Days[numTasks]); |
| } |
| |
| // Check total tasks |
| assertEquals(numTasks, 8); |
| |
| // Drop the table |
| dropRealtimeTable(tableName); |
| |
| // Check if the task metadata is cleaned up on table deletion |
| verifyTableDelete(realtimeTableName); |
| } |
| |
| @AfterClass |
| public void tearDown() |
| throws Exception { |
| stopMinion(); |
| stopServer(); |
| stopBroker(); |
| stopController(); |
| stopKafka(); |
| stopZk(); |
| FileUtils.deleteDirectory(_tempDir); |
| } |
| } |
