voonhous commented on code in PR #19875: URL: https://github.com/apache/hudi/pull/19875#discussion_r3975343136
########## 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, + Arrays.asList(DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH)), + // --num-days walks back from today, so every partition of this table is out of the window + Arguments.of(null, null, 10L, Collections.emptyList())); + } + + private TableSizeStats.Config statsConfig() { + TableSizeStats.Config cfg = new TableSizeStats.Config(); + cfg.basePath = basePath; + cfg.parallelism = 2; + return cfg; + } + + private void writeOneCommit(String... partitions) { + HoodieWriteConfig writeConfig = getConfigBuilder().build(); + try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) { + String instantTime = WriteClientTestUtils.createNewInstantTime(); + List<HoodieRecord> records = new ArrayList<>(); + for (String partition : partitions) { + records.addAll(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); + } + } + + private void writeDefaultPartitions() { + writeOneCommit(DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH); + } + + private List<String> runAndCollectLogs(TableSizeStats.Config cfg) { + CapturingAppender appender = new CapturingAppender(); + Logger logger = (Logger) LogManager.getLogger(TableSizeStats.class); + Level previousLevel = logger.getLevel(); + try { + appender.start(); + logger.setLevel(Level.INFO); Review Comment: Confirmed, and it hides in CI as a flake: with `log4j2-surefire-quiet.properties` the old order captured nothing on the first attempt and passed on surefire's rerun because the second attempt reused the `LoggerConfig` the first one created. The shared appender now sets the level after `addAppender` and restores it on close, and its javadoc says it gets its own `LoggerConfig`. 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, + Arrays.asList(DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH)), + // --num-days walks back from today, so every partition of this table is out of the window + Arguments.of(null, null, 10L, Collections.emptyList())); + } + + private TableSizeStats.Config statsConfig() { + TableSizeStats.Config cfg = new TableSizeStats.Config(); + cfg.basePath = basePath; + cfg.parallelism = 2; + return cfg; + } + + private void writeOneCommit(String... partitions) { + HoodieWriteConfig writeConfig = getConfigBuilder().build(); + try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) { + String instantTime = WriteClientTestUtils.createNewInstantTime(); + List<HoodieRecord> records = new ArrayList<>(); + for (String partition : partitions) { + records.addAll(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); + } + } + + private void writeDefaultPartitions() { + writeOneCommit(DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH); + } + + private List<String> runAndCollectLogs(TableSizeStats.Config cfg) { + CapturingAppender appender = new CapturingAppender(); + Logger logger = (Logger) LogManager.getLogger(TableSizeStats.class); + Level previousLevel = logger.getLevel(); + try { + appender.start(); + logger.setLevel(Level.INFO); + logger.addAppender(appender); + new TableSizeStats(jsc, cfg).run(); + } finally { + logger.removeAppender(appender); + logger.setLevel(previousLevel); + } + return appender.messages(); + } + + private static Set<String> partitionStatHeaders(List<String> messages) { + return messages.stream().filter(m -> m.startsWith(PARTITION_STATS_PREFIX)).collect(Collectors.toSet()); + } + + private static String lineAfter(List<String> messages, String header) { + int index = messages.indexOf(header); + assertTrue(index >= 0 && index + 1 < messages.size(), "missing log line [" + header + "] in " + messages); + return messages.get(index + 1); + } + + @Test + public void testTableAndPartitionStatsCoverEveryPartition() { + writeDefaultPartitions(); + TableSizeStats.Config cfg = statsConfig(); + cfg.tableStats = true; + cfg.partitionStats = true; + + List<String> messages = runAndCollectLogs(cfg); + + Set<String> expectedHeaders = Stream.of( + DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH, DEFAULT_THIRD_PARTITION_PATH) + .map(p -> PARTITION_STATS_PREFIX + p + "]").collect(Collectors.toSet()); + assertEquals(expectedHeaders, partitionStatHeaders(messages)); + for (String header : expectedHeaders) { + assertEquals("Number of files: 1", lineAfter(messages, header)); + } + + String tableHeader = "Table stats [path: " + basePath + "]"; + assertEquals("Number of files: 3", lineAfter(messages, tableHeader)); + assertTrue(messages.stream().anyMatch(m -> m.matches("Total size: \\d+\\.\\d{2} (B|KB|MB|GB|TB)")), + "expected a formatted total size in " + messages); + } + + @Test + public void testTotalSizeOnlyWhenTableStatsAreOff() { + writeDefaultPartitions(); + List<String> messages = runAndCollectLogs(statsConfig()); + + assertEquals(Collections.emptySet(), partitionStatHeaders(messages), + "partition stats must stay off unless asked for"); + assertTrue(messages.stream().noneMatch(m -> m.startsWith("Table stats [path: "))); + assertTrue(messages.stream().anyMatch(m -> m.matches("Total size: \\d+\\.\\d{2} (B|KB|MB|GB|TB)")), + "expected a formatted total size in " + messages); + } + + @ParameterizedTest + @MethodSource("dateIntervalArgs") + public void testOnlyPartitionsInsideTheDateIntervalAreCounted(String startDate, String endDate, long numDays, + List<String> expectedPartitions) { + writeDefaultPartitions(); + TableSizeStats.Config cfg = statsConfig(); + cfg.partitionStats = true; + cfg.startDate = startDate; + cfg.endDate = endDate; + cfg.numDays = numDays; + + List<String> messages = runAndCollectLogs(cfg); + + Set<String> expectedHeaders = expectedPartitions.stream() + .map(p -> PARTITION_STATS_PREFIX + p + ", has date: yes]").collect(Collectors.toSet()); + assertEquals(expectedHeaders, partitionStatHeaders(messages)); + } + + @Test + public void testBasePathsAreReadFromThePropsFile() throws IOException { + writeDefaultPartitions(); + Path propsFile = tempDir.resolve("base-paths.properties"); + Files.write(propsFile, Collections.singletonList(basePath), StandardCharsets.UTF_8); + + TableSizeStats.Config cfg = statsConfig(); + cfg.basePath = null; + cfg.propsFilePath = propsFile.toAbsolutePath().toString(); + cfg.tableStats = true; + + List<String> messages = runAndCollectLogs(cfg); + assertEquals("Number of files: 3", lineAfter(messages, "Table stats [path: " + basePath + "]")); + } + + @Test + public void testUnreadablePropsFileFails() { + TableSizeStats.Config cfg = statsConfig(); + cfg.propsFilePath = tempDir.resolve("missing-" + UUID.randomUUID() + ".properties").toAbsolutePath().toString(); + assertThrows(HoodieException.class, () -> new TableSizeStats(jsc, cfg).run()); Review Comment: Right, the constructor reads the file first, so `getFilePaths`'s catch is unreachable unless the file disappears between construction and `run()`. Renamed to `testMissingPropsFileFailsInTheConstructor` asserting `Properties file does not exist`, and added `testPropsFileRemovedAfterTheConstructorFailsTheRun`, which deletes the file after construction and asserts `Cannot read properties from dfs from file`. 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]
