wombatu-kun commented on code in PR #19875:
URL: https://github.com/apache/hudi/pull/19875#discussion_r3974955121
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieDataTableValidator.java:
##########
@@ -212,7 +212,7 @@ public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) {
return false;
}
- HoodieMetadataTableValidator.Config config =
(HoodieMetadataTableValidator.Config) o;
+ Config config = (Config) o;
Review Comment:
`basePath.equals(config.basePath)` is the one comparison in this method not
routed through `Objects.equals`, so `equals` still throws for a `Config` left
at the default null `basePath`. Worth converting it here, with a
default-instance case in `testConfigEqualsHashCodeAndToString`.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDataTableValidator.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieValidationException;
+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.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+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.assertDoesNotThrow;
+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 HoodieDataTableValidator} against a small three-partition COW
table, with and without
+ * data files that the timeline does not account for.
+ */
+public class TestHoodieDataTableValidator extends HoodieSparkClientTestBase {
+
+ private static final int RECORDS_PER_PARTITION = 4;
+
+ private HoodieDataTableValidator.Config validatorConfig(boolean
ignoreFailed) {
+ HoodieDataTableValidator.Config cfg = new
HoodieDataTableValidator.Config();
+ cfg.basePath = basePath;
+ cfg.parallelism = 2;
+ cfg.ignoreFailed = ignoreFailed;
+ return cfg;
+ }
+
+ private String writeOneCommit() {
+ HoodieWriteConfig writeConfig = getConfigBuilder().build();
+ try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
+ String instantTime = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> records = new ArrayList<>();
+ for (String partition : Arrays.asList(
+ DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH,
DEFAULT_THIRD_PARTITION_PATH)) {
+ 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);
+ return instantTime;
+ }
+ }
+
+ /**
+ * Copies an existing base file of the first partition to a new base file
named after {@code instantTime} and a
+ * brand new file id, which is exactly the shape of a data file the timeline
does not account for.
+ */
+ private void addUnaccountedBaseFile(String instantTime) throws IOException {
+ Path partitionDir = Paths.get(basePath, DEFAULT_FIRST_PARTITION_PATH);
+ Path source;
+ try (Stream<Path> files = Files.list(partitionDir)) {
+ source = files.filter(p -> p.toString().endsWith(".parquet")).findFirst()
+ .orElseThrow(() -> new IllegalStateException("no base file written
under " + partitionDir));
+ }
+ String danglingName =
+ FSUtils.makeBaseFileName(instantTime, "1-0-1",
UUID.randomUUID().toString(), ".parquet");
+ Files.copy(source, partitionDir.resolve(danglingName));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testValidationPassesOnAHealthyTable(boolean
readPropsFromFileSystem) throws IOException {
+ writeOneCommit();
+ HoodieDataTableValidator.Config cfg = validatorConfig(false);
+ if (readPropsFromFileSystem) {
+ Path propsFile = tempDir.resolve("validator.properties");
+ Files.write(propsFile,
+ Collections.singletonList(HoodieWriteConfig.TBL_NAME.key() + "=" +
metaClient.getTableConfig().getTableName()),
+ StandardCharsets.UTF_8);
+ cfg.propsFilePath = propsFile.toAbsolutePath().toString();
Review Comment:
`HoodieDataTableValidator` assigns `props` in its constructor and never
reads it again, so this arm cannot fail differently from the `false` one. Is
`--props` meant to be a no-op for this tool, or is the unread field the real
thing to fix?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDataTableValidator.java:
##########
@@ -0,0 +1,223 @@
+/*
+ * 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.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieValidationException;
+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.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+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.assertDoesNotThrow;
+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 HoodieDataTableValidator} against a small three-partition COW
table, with and without
+ * data files that the timeline does not account for.
+ */
+public class TestHoodieDataTableValidator extends HoodieSparkClientTestBase {
+
+ private static final int RECORDS_PER_PARTITION = 4;
+
+ private HoodieDataTableValidator.Config validatorConfig(boolean
ignoreFailed) {
+ HoodieDataTableValidator.Config cfg = new
HoodieDataTableValidator.Config();
+ cfg.basePath = basePath;
+ cfg.parallelism = 2;
+ cfg.ignoreFailed = ignoreFailed;
+ return cfg;
+ }
+
+ private String writeOneCommit() {
+ HoodieWriteConfig writeConfig = getConfigBuilder().build();
+ try (SparkRDDWriteClient client = getHoodieWriteClient(writeConfig)) {
+ String instantTime = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> records = new ArrayList<>();
+ for (String partition : Arrays.asList(
+ DEFAULT_FIRST_PARTITION_PATH, DEFAULT_SECOND_PARTITION_PATH,
DEFAULT_THIRD_PARTITION_PATH)) {
+ 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);
+ return instantTime;
+ }
+ }
+
+ /**
+ * Copies an existing base file of the first partition to a new base file
named after {@code instantTime} and a
+ * brand new file id, which is exactly the shape of a data file the timeline
does not account for.
+ */
+ private void addUnaccountedBaseFile(String instantTime) throws IOException {
+ Path partitionDir = Paths.get(basePath, DEFAULT_FIRST_PARTITION_PATH);
+ Path source;
+ try (Stream<Path> files = Files.list(partitionDir)) {
+ source = files.filter(p -> p.toString().endsWith(".parquet")).findFirst()
+ .orElseThrow(() -> new IllegalStateException("no base file written
under " + partitionDir));
+ }
+ String danglingName =
+ FSUtils.makeBaseFileName(instantTime, "1-0-1",
UUID.randomUUID().toString(), ".parquet");
+ Files.copy(source, partitionDir.resolve(danglingName));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testValidationPassesOnAHealthyTable(boolean
readPropsFromFileSystem) throws IOException {
+ writeOneCommit();
+ HoodieDataTableValidator.Config cfg = validatorConfig(false);
+ if (readPropsFromFileSystem) {
+ Path propsFile = tempDir.resolve("validator.properties");
+ Files.write(propsFile,
+ Collections.singletonList(HoodieWriteConfig.TBL_NAME.key() + "=" +
metaClient.getTableConfig().getTableName()),
+ StandardCharsets.UTF_8);
+ cfg.propsFilePath = propsFile.toAbsolutePath().toString();
+ }
+ HoodieDataTableValidator validator = new HoodieDataTableValidator(jsc,
cfg);
+ // the validator reports through an exception only, so a clean table is
asserted by the absence of one
+ assertDoesNotThrow(validator::run);
+ }
+
+ @Test
+ public void testMissingPropsFileFails() {
+ HoodieDataTableValidator.Config cfg = validatorConfig(false);
+ cfg.propsFilePath =
tempDir.resolve("does-not-exist.properties").toAbsolutePath().toString();
+ assertThrows(HoodieIOException.class, () -> new
HoodieDataTableValidator(jsc, cfg));
+ }
+
+ /**
+ * A base file whose instant time precedes the first instant of the active
timeline is dangling; whether that
+ * fails the job depends on --ignore-failed.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testDanglingFileBeforeTheActiveTimeline(boolean ignoreFailed)
throws IOException {
+ writeOneCommit();
+ addUnaccountedBaseFile("00000000000001");
+
+ HoodieDataTableValidator validator = new HoodieDataTableValidator(jsc,
validatorConfig(ignoreFailed));
+ if (ignoreFailed) {
+ assertDoesNotThrow(validator::run);
Review Comment:
With `--ignore-failed` on, `doDataTableValidation` only logs and sets
`finalResult`, so this arm passes whether the dangling file was detected and
ignored or never detected at all. Capturing the validator's ERROR line, the way
`TestTableSizeStats` captures the tool's log, would let the arm fail.
##########
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");
Review Comment:
Every assertion here shows only that nothing changed, so what dry-run mode
actually reports through `printDeleteFilesInfo` is never checked. Asserting the
two named partitions appear in that output would cover the mode rather than its
absence of side effects.
##########
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:
`addAppender` installs a dedicated LoggerConfig for this logger and calls
`updateLoggers`, which discards the preceding `setLevel(Level.INFO)`, so the
capture works only because the surefire config puts `org.apache.hudi` at debug
and returns nothing under `log4j2-surefire-quiet.properties`. Swapping the two
calls makes the level stick, and the `CapturingAppender` javadoc needs the
matching correction - the appender does not land on the shared nearest logger.
##########
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:
`run` wraps every failure as `HoodieException(..., e)`, so the cause is
non-null by construction and this holds even if the tool failed before hive
sync was reached. Asserting the cause's type or message, as the sibling test
does for `--hive-database`, would tie it to the sync.
##########
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");
Review Comment:
Both hive tests pin `verifyHiveConfigs` running only after
`doDeleteTablePartitions` has committed, so an operator who typos
`--hive-database` loses the partitions before the error surfaces. Is freezing
that order intended, or should the hive check move to the top of `run`?
##########
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:
No partition sits on either bound here, so this case selects the same two
partitions as the previous one and proves nothing about `[start, end)`. A
`2015/3/16` start (must include the second partition) and a `2015/3/17` end
(must exclude the third) would pin the half-open behaviour the comment claims.
##########
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:
The constructor already parses `--props-path` through
`readConfigFromFileSystem`, so this throws before `run` is entered and
`getFilePaths`'s own read-failure branch stays uncovered. Asserting the message
would show which of the two reads actually failed.
--
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]
