voonhous commented on code in PR #19875: URL: https://github.com/apache/hudi/pull/19875#discussion_r3975574215
########## hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDropPartitionsTool.java: ########## @@ -0,0 +1,282 @@ +/* + * 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.hudi.utilities; + +import org.apache.hudi.client.SparkRDDWriteClient; +import org.apache.hudi.client.WriteClientTestUtils; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieReplaceCommitMetadata; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.testutils.HoodieSparkClientTestBase; + +import org.apache.spark.api.java.JavaRDD; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.stream.Collectors; + +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH; +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH; +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests {@link HoodieDropPartitionsTool} against a small three-partition COW table. + */ +public class TestHoodieDropPartitionsTool extends HoodieSparkClientTestBase { + + private static final int RECORDS_PER_PARTITION = 4; + + private HoodieDropPartitionsTool.Config toolConfig(String mode, String partitions) { + HoodieDropPartitionsTool.Config cfg = new HoodieDropPartitionsTool.Config(); + cfg.basePath = basePath; + cfg.tableName = metaClient.getTableConfig().getTableName(); + cfg.runningMode = mode; + cfg.partitions = partitions; + cfg.parallelism = 2; + cfg.configs.add(HoodieWriteConfig.TBL_NAME.key() + "=" + cfg.tableName); + return cfg; + } + + /** + * Writes two insert commits: the first spreads records over all three partitions, the second adds a + * second file slice to the first partition. + */ + private void writeThreePartitionTable() { + HoodieWriteConfig writeConfig = getConfigBuilder().build(); + try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) { + String firstCommit = WriteClientTestUtils.createNewInstantTime(); + List<HoodieRecord> firstBatch = new ArrayList<>(); + for (String partition : Arrays.asList( + DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH)) { + firstBatch.addAll(dataGen.generateInsertsForPartition(firstCommit, RECORDS_PER_PARTITION, partition)); + } + writeBatchAndCommit(client, firstCommit, firstBatch); + + String secondCommit = WriteClientTestUtils.createNewInstantTime(); + writeBatchAndCommit(client, secondCommit, + dataGen.generateInsertsForPartition(secondCommit, RECORDS_PER_PARTITION, DEFAULT_FIRST_PARTITION_PATH)); + } + } + + private void writeBatchAndCommit(SparkRDDWriteClient client, String instantTime, List<HoodieRecord> records) { + WriteClientTestUtils.startCommitWithTime(client, instantTime); + JavaRDD<WriteStatus> writeStatuses = client.insert(jsc.parallelize(records, 1), instantTime); + client.commit(instantTime, writeStatuses); + } + + private long latestBaseFileCount(String partition) { + HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient); + try (HoodieTableFileSystemView fsView = FileSystemViewManager.createInMemoryFileSystemView( + context, reloaded, HoodieMetadataConfig.newBuilder().enable(false).build())) { + return fsView.getLatestBaseFiles(partition).count(); + } + } + + private List<String> completedInstants() { + return HoodieTableMetaClient.reload(metaClient).getActiveTimeline().filterCompletedInstants() + .getInstantsAsStream().map(HoodieInstant::requestedTime).collect(Collectors.toList()); + } + + @Test + public void testDryRunLeavesTableUntouched() { + writeThreePartitionTable(); + List<String> instantsBefore = completedInstants(); + + HoodieDropPartitionsTool.Config cfg = toolConfig("dry_run", + DEFAULT_FIRST_PARTITION_PATH + "," + DEFAULT_SECOND_PARTITION_PATH); + new HoodieDropPartitionsTool(jsc, cfg).run(); + + assertEquals(instantsBefore, completedInstants(), "dry run must not add any instant"); + assertEquals(1, latestBaseFileCount(DEFAULT_FIRST_PARTITION_PATH)); + assertEquals(1, latestBaseFileCount(DEFAULT_SECOND_PARTITION_PATH)); + assertEquals(1, latestBaseFileCount(DEFAULT_THIRD_PARTITION_PATH)); + } + + @Test + public void testDeleteMasksOnlyTheRequestedPartitions() throws IOException { + writeThreePartitionTable(); + int instantsBefore = completedInstants().size(); + + HoodieDropPartitionsTool.Config cfg = toolConfig("delete", + DEFAULT_FIRST_PARTITION_PATH + "," + DEFAULT_SECOND_PARTITION_PATH); + new HoodieDropPartitionsTool(jsc, cfg).run(); + + HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient); + assertEquals(instantsBefore + 1, completedInstants().size(), "delete must add exactly one instant"); + HoodieInstant replaceInstant = reloaded.getActiveTimeline().getCompletedReplaceTimeline().lastInstant().get(); + HoodieReplaceCommitMetadata replaceMetadata = + reloaded.getActiveTimeline().readReplaceCommitMetadata(replaceInstant); + assertEquals( + new HashSet<>(Arrays.asList(DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH)), + replaceMetadata.getPartitionToReplaceFileIds().keySet()); + // the file group of the first partition, written by both commits, is masked + assertEquals(1, replaceMetadata.getPartitionToReplaceFileIds().get(DEFAULT_FIRST_PARTITION_PATH).size()); + + assertEquals(0, latestBaseFileCount(DEFAULT_FIRST_PARTITION_PATH)); + assertEquals(0, latestBaseFileCount(DEFAULT_SECOND_PARTITION_PATH)); + assertEquals(1, latestBaseFileCount(DEFAULT_THIRD_PARTITION_PATH), + "the partition that was not named must survive"); + } + + /** + * The tool takes its write properties either from --props or from repeated --hoodie-conf, and only defaults + * hoodie.meta.fields.mode from the table when the operator did not name it. Both sources are checked by asking + * for a meta-fields mode the table does not have and expecting the write config gate to reject it. + */ + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testWritePropertiesComeFromPropsFileAndHoodieConf(boolean usePropsFile) throws IOException { + writeThreePartitionTable(); + + HoodieDropPartitionsTool.Config cfg = toolConfig("dry_run", DEFAULT_THIRD_PARTITION_PATH); + String metaFieldsOverride = "hoodie.meta.fields.mode=NONE"; + if (usePropsFile) { + // the file carries the mode, the --hoodie-conf entry already on the config carries the table name, so + // both sources have to be merged for this run to reach the write config gate + Path propsFile = tempDir.resolve("drop-partitions.properties"); + Files.write(propsFile, Collections.singletonList(metaFieldsOverride), StandardCharsets.UTF_8); + cfg.propsFilePath = propsFile.toAbsolutePath().toString(); + } else { + cfg.configs.add(metaFieldsOverride); + } + + HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg); + Throwable thrown = assertThrows(HoodieException.class, tool::run); + assertTrue(stackMessages(thrown).contains("hoodie.meta.fields.mode"), + "expected the meta fields mode from the config source to reach the write config, got: " + thrown); + } + + @Test + public void testUnsupportedModeFails() { + writeThreePartitionTable(); + HoodieDropPartitionsTool.Config cfg = toolConfig("purge", DEFAULT_THIRD_PARTITION_PATH); + HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg); + + HoodieException thrown = assertThrows(HoodieException.class, tool::run); + assertTrue(thrown.getMessage().contains("Unable to delete table partitions in " + basePath)); + assertTrue(thrown.getCause() instanceof IllegalArgumentException, "got " + thrown.getCause()); + assertEquals(0, HoodieTableMetaClient.reload(metaClient).getActiveTimeline() + .getCompletedReplaceTimeline().countInstants()); + } + + /** + * Hive sync is verified after the partitions have already been masked, so a missing --hive-database fails the + * job even though the drop itself is committed. + */ + @Test + public void testHiveSyncWithoutDatabaseFailsAfterTheDrop() { + writeThreePartitionTable(); + HoodieDropPartitionsTool.Config cfg = toolConfig("delete", DEFAULT_THIRD_PARTITION_PATH); + cfg.syncToHive = true; + cfg.hiveDataBase = null; + HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg); + + HoodieException thrown = assertThrows(HoodieException.class, tool::run); + assertTrue(thrown.getCause() instanceof IllegalArgumentException, "got " + thrown.getCause()); + assertTrue(thrown.getCause().getMessage().contains("--hive-database")); + assertEquals(1, HoodieTableMetaClient.reload(metaClient).getActiveTimeline() + .getCompletedReplaceTimeline().countInstants(), "the partitions are dropped before hive sync runs"); + assertEquals(0, latestBaseFileCount(DEFAULT_THIRD_PARTITION_PATH)); + } + + /** + * With the hive configs in place the sync props are built and the sync itself is attempted; pointing it at a + * port nothing listens on keeps the test free of a metastore while still running that path. + */ + @Test + public void testHiveSyncFailureLeavesTheDropCommitted() { + writeThreePartitionTable(); + HoodieDropPartitionsTool.Config cfg = toolConfig("delete", DEFAULT_THIRD_PARTITION_PATH); + cfg.syncToHive = true; + cfg.hiveDataBase = "db"; + cfg.hiveTableName = "tbl"; + cfg.hivePartitionsField = "partition_path"; + cfg.hiveHMSUris = "thrift://localhost:1"; + HoodieDropPartitionsTool tool = new HoodieDropPartitionsTool(jsc, cfg); + + HoodieException thrown = assertThrows(HoodieException.class, tool::run); + assertNotNull(thrown.getCause(), "the hive sync failure must be reported as the cause"); + assertEquals(1, HoodieTableMetaClient.reload(metaClient).getActiveTimeline() + .getCompletedReplaceTimeline().countInstants(), "the drop is committed before hive sync runs"); + assertEquals(0, latestBaseFileCount(DEFAULT_THIRD_PARTITION_PATH)); + } + + @Test + public void testConfigEqualsHashCodeAndToString() { + HoodieDropPartitionsTool.Config left = new HoodieDropPartitionsTool.Config(); + left.basePath = "/tmp/table"; + left.runningMode = "delete"; + left.tableName = "t1"; + left.partitions = "p1,p2"; + left.configs = new ArrayList<>(Collections.singletonList("k=v")); + + HoodieDropPartitionsTool.Config right = new HoodieDropPartitionsTool.Config(); + right.basePath = "/tmp/table"; + right.runningMode = "delete"; + right.tableName = "t1"; + right.partitions = "p1,p2"; + right.configs = new ArrayList<>(Collections.singletonList("k=v")); + + assertEquals(left, left); + assertEquals(left, right); + assertEquals(left.hashCode(), right.hashCode()); + assertNotEquals(left, null); + assertNotEquals(left, "not a config"); + + right.hiveDataBase = "db"; + assertNotEquals(left, right); + assertNotEquals(left.hashCode(), right.hashCode()); + + String printed = left.toString(); + assertTrue(printed.contains("--base-path /tmp/table")); + assertTrue(printed.contains("--partitions p1,p2")); + assertTrue(printed.contains("--hoodie-conf [k=v]")); + assertTrue(printed.contains("--hive-user-name Masked"), "credentials must not be printed"); + } + + private static String stackMessages(Throwable throwable) { Review Comment: Done: `stackMessages` lives once in a shared `ToolTestUtils` next to `CapturingLogAppender` and the two copies (drop tool and data validator; `TestTableSizeStats` never had one) are gone. Done in 53af3073ee38. ########## hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieTTLJob.java: ########## @@ -0,0 +1,140 @@ +/* + * 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.hudi.utilities; + +import org.apache.hudi.client.SparkRDDWriteClient; +import org.apache.hudi.client.WriteClientTestUtils; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieReplaceCommitMetadata; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.view.FileSystemViewManager; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.config.HoodieCleanConfig; +import org.apache.hudi.config.HoodieTTLConfig; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.testutils.HoodieSparkClientTestBase; + +import org.apache.spark.api.java.JavaRDD; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; + +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH; +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH; +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_THIRD_PARTITION_PATH; +import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.getCommitTimeAtUTC; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; + +/** + * Tests {@link HoodieTTLJob} on a table whose partitions were written at different times. + */ +public class TestHoodieTTLJob extends HoodieSparkClientTestBase { + + private static final int RECORDS_PER_PARTITION = 4; + + /** + * Both constructors are covered: with an explicit props/meta client pair, and with the (jsc, cfg) constructor + * that has to read --props and --hoodie-conf itself. + */ + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testTtlDropsOnlyExpiredPartitions(boolean readPropsFromFileSystem) throws IOException { + writeOnePartitionPerInstant(); + + HoodieTTLJob.Config cfg = new HoodieTTLJob.Config(); + cfg.basePath = basePath; + cfg.parallelism = 2; + + HoodieTTLJob job; + if (readPropsFromFileSystem) { + Path propsFile = tempDir.resolve("ttl.properties"); + Files.write(propsFile, Arrays.asList( + HoodieWriteConfig.TBL_NAME.key() + "=" + metaClient.getTableConfig().getTableName(), + HoodieTTLConfig.PARTITION_TTL_STRATEGY_TYPE.key() + "=KEEP_BY_TIME"), StandardCharsets.UTF_8); + cfg.propsFilePath = propsFile.toAbsolutePath().toString(); + cfg.configs.add(HoodieTTLConfig.DAYS_RETAIN.key() + "=10"); + job = new HoodieTTLJob(jsc, cfg); + } else { + TypedProperties props = new TypedProperties(); + props.setProperty(HoodieWriteConfig.TBL_NAME.key(), metaClient.getTableConfig().getTableName()); + props.setProperty(HoodieTTLConfig.PARTITION_TTL_STRATEGY_TYPE.key(), "KEEP_BY_TIME"); + props.setProperty(HoodieTTLConfig.DAYS_RETAIN.key(), "10"); + job = new HoodieTTLJob(jsc, cfg, props, metaClient); + // the job turns async cleaning off on the properties it was handed + assertEquals("false", props.get(HoodieCleanConfig.ASYNC_CLEAN.key()).toString()); + } + + job.run(); + + HoodieTableMetaClient reloaded = HoodieTableMetaClient.reload(metaClient); + HoodieInstant replaceInstant = + reloaded.getActiveTimeline().getCompletedReplaceTimeline().lastInstant().get(); + HoodieReplaceCommitMetadata replaceMetadata = + reloaded.getActiveTimeline().readReplaceCommitMetadata(replaceInstant); + assertEquals( + new HashSet<>(Arrays.asList(DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH)), + replaceMetadata.getPartitionToReplaceFileIds().keySet(), + "only the partitions older than the retention are dropped"); + + assertEquals(0, latestBaseFileCount(DEFAULT_FIRST_PARTITION_PATH)); + assertEquals(0, latestBaseFileCount(DEFAULT_SECOND_PARTITION_PATH)); + assertEquals(1, latestBaseFileCount(DEFAULT_THIRD_PARTITION_PATH), + "the fresh partition must survive"); + assertFalse(replaceMetadata.getPartitionToReplaceFileIds().containsKey(DEFAULT_THIRD_PARTITION_PATH)); + } + + private void writeOnePartitionPerInstant() { + HoodieWriteConfig writeConfig = getConfigBuilder().build(); + try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) { + // two partitions written far in the past, one written now + writeRecordsForPartition(client, DEFAULT_FIRST_PARTITION_PATH, getCommitTimeAtUTC(0)); + writeRecordsForPartition(client, DEFAULT_SECOND_PARTITION_PATH, getCommitTimeAtUTC(1000)); + writeRecordsForPartition(client, DEFAULT_THIRD_PARTITION_PATH, WriteClientTestUtils.createNewInstantTime()); + } + } + + private void writeRecordsForPartition(SparkRDDWriteClient client, String partition, String instantTime) { + List<HoodieRecord> records = + new ArrayList<>(dataGen.generateInsertsForPartition(instantTime, RECORDS_PER_PARTITION, partition)); + WriteClientTestUtils.startCommitWithTime(client, instantTime); + JavaRDD<WriteStatus> writeStatuses = client.insert(jsc.parallelize(records, 1), instantTime); + client.commit(instantTime, writeStatuses); + } Review Comment: Done: `latestBaseFileCount` moved into the same `ToolTestUtils` as a static helper taking the engine context and meta client; both classes call it. Done in 53af3073ee38. -- 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]
