This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/pinot-ingestion-tests-consolidate-704898 in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 86d6b184a8b9beccb246124c88c3a1f41a6e2810 Author: Xiang Fu <[email protected]> AuthorDate: Fri Aug 7 19:46:44 2026 -0700 Fold the metadata-push purge test into PurgeMinionClusterIntegrationTest PurgeMetadataPushMinionClusterIntegrationTest stood up a whole cluster (ZK, controller, broker, server, minion, Kafka) to run a single test. It extended PurgeMinionClusterIntegrationTest and disabled five inherited tests with @Test(enabled = false), which suppresses the assertions but not the setup: of its 118.7s per CI run, 106.6s was @BeforeClass. It is now a seventh offline table plus one extra test method on the parent, sharing the cluster the parent already builds. The first-run purge flow is extracted into runFirstRunPurge(tableName) so the default push mode and METADATA push mode reuse it. The new table is inserted before PURGE_ALL_RECORDS_TABLE in the setup loop so the trailing tableConfig/schema that buildSegmentsFromAvro consumes is unchanged. Measured locally: 7 tests in 117s, against 159.0s + 118.7s across two cluster startups before. --- ...geMetadataPushMinionClusterIntegrationTest.java | 73 ---------------------- .../tests/PurgeMinionClusterIntegrationTest.java | 50 ++++++++++++--- 2 files changed, 40 insertions(+), 83 deletions(-) diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMetadataPushMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMetadataPushMinionClusterIntegrationTest.java deleted file mode 100644 index 2f2057e3970..00000000000 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMetadataPushMinionClusterIntegrationTest.java +++ /dev/null @@ -1,73 +0,0 @@ -/** - * 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 java.util.HashMap; -import java.util.Map; -import org.apache.pinot.core.common.MinionConstants; -import org.apache.pinot.plugin.minion.tasks.MinionTaskUtils; -import org.apache.pinot.spi.config.table.TableTaskConfig; -import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties; -import org.testng.annotations.Test; - - -/// Integration test that runs the Purge minion task with METADATA -/// [org.apache.pinot.spi.ingestion.batch.BatchConfigProperties.SegmentPushType] to verify the full flow. -/// Only [#testFirstRunPurge()] is enabled; other tests from the base class are disabled. -public class PurgeMetadataPushMinionClusterIntegrationTest extends PurgeMinionClusterIntegrationTest { - - @Override - protected TableTaskConfig getPurgeTaskConfig() { - Map<String, String> tableTaskConfigs = new HashMap<>(); - tableTaskConfigs.put(MinionConstants.PurgeTask.LAST_PURGE_TIME_THREESOLD_PERIOD, "1d"); - tableTaskConfigs.put(BatchConfigProperties.PUSH_MODE, BatchConfigProperties.SegmentPushType.METADATA.name()); - tableTaskConfigs.put(MinionTaskUtils.ALLOW_METADATA_PUSH_WITH_LOCAL_FS, "true"); - return new TableTaskConfig(Map.of(MinionConstants.PurgeTask.TASK_TYPE, tableTaskConfigs)); - } - - @Override - @Test(enabled = false) - public void testPassedDelayTimePurge() { - // Disabled: only testFirstRunPurge runs for METADATA push flow. - } - - @Override - @Test(enabled = false) - public void testNotPassedDelayTimePurge() { - // Disabled: only testFirstRunPurge runs for METADATA push flow. - } - - @Override - @Test(enabled = false) - public void testPurgeOnOldSegmentsWithIndicesOnNewColumns() { - // Disabled: only testFirstRunPurge runs for METADATA push flow. - } - - @Override - @Test(enabled = false) - public void testSegmentDeletionWhenAllRecordsPurged() { - // Disabled: only testFirstRunPurge runs for METADATA push flow. - } - - @Override - @Test(enabled = false) - public void testRealtimeLastSegmentPreservation() { - // Disabled: only testFirstRunPurge runs for METADATA push flow. - } -} diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java index 6dd6a05a6df..71ceaf41204 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PurgeMinionClusterIntegrationTest.java @@ -20,7 +20,6 @@ package org.apache.pinot.integration.tests; import java.io.File; import java.util.ArrayList; -import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -36,12 +35,14 @@ import org.apache.pinot.controller.helix.core.minion.PinotTaskManager; import org.apache.pinot.controller.helix.core.minion.TaskSchedulingContext; import org.apache.pinot.core.common.MinionConstants; import org.apache.pinot.minion.MinionContext; +import org.apache.pinot.plugin.minion.tasks.MinionTaskUtils; import org.apache.pinot.spi.config.table.IndexingConfig; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.TableTaskConfig; import org.apache.pinot.spi.data.DimensionFieldSpec; import org.apache.pinot.spi.data.FieldSpec; import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties; import org.apache.pinot.spi.utils.builder.TableNameBuilder; import org.apache.pinot.util.TestUtils; import org.testng.annotations.AfterClass; @@ -61,6 +62,14 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes private static final String PURGE_OLD_SEGMENTS_WITH_NEW_INDICES_TABLE = "myTable4"; private static final String PURGE_ALL_RECORDS_TABLE = "myTable5"; private static final String PURGE_REALTIME_LAST_SEGMENT_TABLE = "myTable6"; + /// Same first-run purge flow as [#PURGE_FIRST_RUN_TABLE], but pushing the purged segments back + /// with [BatchConfigProperties.SegmentPushType#METADATA]. + private static final String PURGE_METADATA_PUSH_TABLE = "myTable7"; + + /// Tables whose record purger drops the 52 rows with `ArrTime = 1`, leaving 115493 of 115545. + private static final Set<String> ARR_TIME_PURGE_TABLES = + Set.of(PURGE_FIRST_RUN_TABLE, PURGE_DELTA_PASSED_TABLE, PURGE_DELTA_NOT_PASSED_TABLE, + PURGE_OLD_SEGMENTS_WITH_NEW_INDICES_TABLE, PURGE_METADATA_PUSH_TABLE); protected PinotHelixTaskResourceManager _helixTaskResourceManager; protected PinotTaskManager _taskManager; @@ -87,7 +96,7 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes List<String> allOfflineTables = List.of(PURGE_FIRST_RUN_TABLE, PURGE_DELTA_PASSED_TABLE, PURGE_DELTA_NOT_PASSED_TABLE, - PURGE_OLD_SEGMENTS_WITH_NEW_INDICES_TABLE, PURGE_ALL_RECORDS_TABLE); + PURGE_OLD_SEGMENTS_WITH_NEW_INDICES_TABLE, PURGE_METADATA_PUSH_TABLE, PURGE_ALL_RECORDS_TABLE); Schema schema = null; TableConfig tableConfig = null; for (String tableName : allOfflineTables) { @@ -99,7 +108,8 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes // create and upload table config setTableName(tableName); tableConfig = createOfflineTableConfig(); - tableConfig.setTaskConfig(getPurgeTaskConfig()); + tableConfig.setTaskConfig( + PURGE_METADATA_PUSH_TABLE.equals(tableName) ? getMetadataPushPurgeTaskConfig() : getPurgeTaskConfig()); addTableConfig(tableConfig); } @@ -164,10 +174,7 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes private void setRecordPurger() { MinionContext minionContext = MinionContext.getInstance(); minionContext.setRecordPurgerFactory(rawTableName -> { - List<String> tableNames = - Arrays.asList(PURGE_FIRST_RUN_TABLE, PURGE_DELTA_PASSED_TABLE, PURGE_DELTA_NOT_PASSED_TABLE, - PURGE_OLD_SEGMENTS_WITH_NEW_INDICES_TABLE); - if (tableNames.contains(rawTableName)) { + if (ARR_TIME_PURGE_TABLES.contains(rawTableName)) { return row -> row.getValue("ArrTime").equals(1); } else if (PURGE_ALL_RECORDS_TABLE.equals(rawTableName) || PURGE_REALTIME_LAST_SEGMENT_TABLE.equals( rawTableName)) { @@ -194,10 +201,33 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes return new TableTaskConfig(Map.of(MinionConstants.PurgeTask.TASK_TYPE, tableTaskConfigs)); } + /// Same as [#getPurgeTaskConfig()], but pushes the purged segments back with METADATA push instead of the + /// default segment push. + protected TableTaskConfig getMetadataPushPurgeTaskConfig() { + Map<String, String> tableTaskConfigs = new HashMap<>(); + tableTaskConfigs.put(MinionConstants.PurgeTask.LAST_PURGE_TIME_THREESOLD_PERIOD, "1d"); + tableTaskConfigs.put(BatchConfigProperties.PUSH_MODE, BatchConfigProperties.SegmentPushType.METADATA.name()); + tableTaskConfigs.put(MinionTaskUtils.ALLOW_METADATA_PUSH_WITH_LOCAL_FS, "true"); + return new TableTaskConfig(Map.of(MinionConstants.PurgeTask.TASK_TYPE, tableTaskConfigs)); + } + /// Test purge with no metadata on the segments (checking null safe implementation) @Test public void testFirstRunPurge() throws Exception { + runFirstRunPurge(PURGE_FIRST_RUN_TABLE); + } + + /// Same first-run purge flow as [#testFirstRunPurge()], but against a table whose purge task pushes the + /// rewritten segments back with METADATA push. Shares this class's cluster rather than standing up its own. + @Test + public void testFirstRunPurgeWithMetadataPush() + throws Exception { + runFirstRunPurge(PURGE_METADATA_PUSH_TABLE); + } + + private void runFirstRunPurge(String tableName) + throws Exception { // Expected purge task generation : // 1. No previous purge run so all segment should be processed and purge metadata should be added to the segments // 2. Check that we cannot run on same time two purge generation ensuring running segment will be skipped @@ -205,7 +235,7 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes // 4. Check after the first run of the purge if we rerun a purge task generation no task should be scheduled // 5. Check the purge process itself by setting an expecting number of rows - String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(PURGE_FIRST_RUN_TABLE); + String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(tableName); assertNotNull(_taskManager.scheduleTasks(new TaskSchedulingContext() .setTablesToSchedule(Set.of(offlineTableName))) .get(MinionConstants.PurgeTask.TASK_TYPE)); @@ -234,11 +264,11 @@ public class PurgeMinionClusterIntegrationTest extends BaseClusterIntegrationTes // 115545 totals rows // Expecting 115545 - 52 = 115493 rows after purging // It might take some time for server to load the purged segments - TestUtils.waitForCondition(aVoid -> getCurrentCountStarResult(PURGE_FIRST_RUN_TABLE) == 115493, 60_000L, + TestUtils.waitForCondition(aVoid -> getCurrentCountStarResult(tableName) == 115493, 60_000L, "Failed to get expected purged records"); // Drop the table - dropOfflineTable(PURGE_FIRST_RUN_TABLE); + dropOfflineTable(tableName); // Check if the task metadata is cleaned up on table deletion verifyTableDelete(offlineTableName); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
