voonhous commented on code in PR #19875: URL: https://github.com/apache/hudi/pull/19875#discussion_r3975341741
########## 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"); Review Comment: Turned out the run never reached the metastore at all: `buildHiveSyncProps` put two `ConfigProperty` objects in as property keys, so `new HiveSyncConfig(props, conf)` threw `ClassCastException` before any connection attempt, which means `--sync-hive-meta` could not work with or without a metastore. Fixed the keys (`.key()`) and made the two boolean values strings. The test now asserts the direct cause is a `HoodieHiveSyncException` and the chain carries `Failed to create HiveMetaStoreClient` and `Could not connect to meta store using any of the URIs provided`; without the fix it fails on the `ClassCastException`. Done in f687e1763e26. ########## hudi-utilities/src/test/java/org/apache/hudi/utilities/TestTableSizeStats.java: ########## @@ -0,0 +1,311 @@ +/* + * 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.model.HoodieRecord; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.testutils.HoodieSparkClientTestBase; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +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.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +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.List; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests {@link TableSizeStats}. The tool reports through its log, so the assertions read back the lines it + * logged for the table and for every partition it decided to include. + */ +public class TestTableSizeStats extends HoodieSparkClientTestBase { + + private static final int RECORDS_PER_PARTITION = 4; + private static final String PARTITION_STATS_PREFIX = "Partition stats [name: "; + + private static Stream<Arguments> dateIntervalArgs() { + return Stream.of( + // only the 2016 partition is on or after the start date + Arguments.of("2016/1/1", null, 0L, Collections.singletonList(DEFAULT_FIRST_PARTITION_PATH)), + // only the 2015 partitions are before the end date + Arguments.of(null, "2016/1/1", 0L, + Arrays.asList(DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH)), + // half open interval [start, end) + Arguments.of("2015/1/1", "2016/1/1", 0L, Review Comment: Done: the case is now start `2015/3/16` and end `2015/3/17`, expecting exactly `2015/03/16` (start included, end excluded, `2016/03/15` out). Done in f687e1763e26. -- 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]
