You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2026/05/15 01:20:42 UTC
(hudi) branch master updated: fix: clear Hive work map after combine split failures (#18719)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 071b3f125b41 fix: clear Hive work map after combine split failures (#18719)
071b3f125b41 is described below
commit 071b3f125b410b3feb8ee90a2ee8073eec67e254
Author: Asish Kumar <87...@users.noreply.github.com>
AuthorDate: Fri May 15 06:50:35 2026 +0530
fix: clear Hive work map after combine split failures (#18719)
* fix: clear Hive work map after combine split failures
HoodieCombineHiveInputFormat cleared Hive's work map only after successful split generation, leaving ThreadLocal work state behind when split classification or generation failed.
Move the cleanup into a finally block and add a regression test that forces getSplits to fail before verifying Utilities.clearWorkMapForConf is still invoked.
---
.../hadoop/hive/HoodieCombineHiveInputFormat.java | 174 +++++++++++----------
.../hive/TestHoodieCombineHiveInputFormat.java | 32 ++++
2 files changed, 120 insertions(+), 86 deletions(-)
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java
index a71695be427e..9634b7f6b097 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java
@@ -324,111 +324,113 @@ public class HoodieCombineHiveInputFormat<K extends WritableComparable, V extend
public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
PerfLogger perfLogger = SessionState.getPerfLogger();
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.GET_SPLITS);
- init(job);
- List<InputSplit> result = new ArrayList<>();
+ try {
+ init(job);
+ List<InputSplit> result = new ArrayList<>();
- Path[] paths = getInputPaths(job);
+ Path[] paths = getInputPaths(job);
- List<Path> nonCombinablePaths = new ArrayList<>(paths.length / 2);
- List<Path> combinablePaths = new ArrayList<>(paths.length / 2);
+ List<Path> nonCombinablePaths = new ArrayList<>(paths.length / 2);
+ List<Path> combinablePaths = new ArrayList<>(paths.length / 2);
- int numThreads = Math.min(MAX_CHECK_NONCOMBINABLE_THREAD_NUM,
- (int) Math.ceil((double) paths.length / DEFAULT_NUM_PATH_PER_THREAD));
+ int numThreads = Math.min(MAX_CHECK_NONCOMBINABLE_THREAD_NUM,
+ (int) Math.ceil((double) paths.length / DEFAULT_NUM_PATH_PER_THREAD));
- // This check is necessary because for Spark branch, the result array from
- // getInputPaths() above could be empty, and therefore numThreads could be 0.
- // In that case, Executors.newFixedThreadPool will fail.
- if (numThreads > 0) {
- try {
- Set<Integer> nonCombinablePathIndices = getNonCombinablePathIndices(job, paths, numThreads);
- for (int i = 0; i < paths.length; i++) {
- if (nonCombinablePathIndices.contains(i)) {
- nonCombinablePaths.add(paths[i]);
- } else {
- combinablePaths.add(paths[i]);
+ // This check is necessary because for Spark branch, the result array from
+ // getInputPaths() above could be empty, and therefore numThreads could be 0.
+ // In that case, Executors.newFixedThreadPool will fail.
+ if (numThreads > 0) {
+ try {
+ Set<Integer> nonCombinablePathIndices = getNonCombinablePathIndices(job, paths, numThreads);
+ for (int i = 0; i < paths.length; i++) {
+ if (nonCombinablePathIndices.contains(i)) {
+ nonCombinablePaths.add(paths[i]);
+ } else {
+ combinablePaths.add(paths[i]);
+ }
}
+ } catch (Exception e) {
+ LOG.error("Error checking non-combinable path", e);
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
+ throw new IOException(e);
}
- } catch (Exception e) {
- LOG.error("Error checking non-combinable path", e);
- perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
- throw new IOException(e);
}
- }
- // Store the previous value for the path specification
- String oldPaths = job.get(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR);
- LOG.debug("The received input paths are: [{}] against the property {}", oldPaths,
- org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR);
-
- // Process the normal splits
- if (nonCombinablePaths.size() > 0) {
- FileInputFormat.setInputPaths(job, nonCombinablePaths.toArray(new Path[0]));
- InputSplit[] splits = super.getSplits(job, numSplits);
- Collections.addAll(result, splits);
- }
+ // Store the previous value for the path specification
+ String oldPaths = job.get(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR);
+ LOG.debug("The received input paths are: [{}] against the property {}", oldPaths,
+ org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR);
- // Process the combine splits
- if (combinablePaths.size() > 0) {
- FileInputFormat.setInputPaths(job, combinablePaths.toArray(new Path[0]));
- Map<Path, PartitionDesc> pathToPartitionInfo = this.pathToPartitionInfo != null ? this.pathToPartitionInfo
- : Utilities.getMapWork(job).getPathToPartitionInfo();
- InputSplit[] splits = getCombineSplits(job, numSplits, pathToPartitionInfo);
- Collections.addAll(result, splits);
- }
+ // Process the normal splits
+ if (nonCombinablePaths.size() > 0) {
+ FileInputFormat.setInputPaths(job, nonCombinablePaths.toArray(new Path[0]));
+ InputSplit[] splits = super.getSplits(job, numSplits);
+ Collections.addAll(result, splits);
+ }
- // Restore the old path information back
- // This is just to prevent incompatibilities with previous versions Hive
- // if some application depends on the original value being set.
- if (oldPaths != null) {
- job.set(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR, oldPaths);
- }
+ // Process the combine splits
+ if (combinablePaths.size() > 0) {
+ FileInputFormat.setInputPaths(job, combinablePaths.toArray(new Path[0]));
+ Map<Path, PartitionDesc> pathToPartitionInfo = this.pathToPartitionInfo != null ? this.pathToPartitionInfo
+ : Utilities.getMapWork(job).getPathToPartitionInfo();
+ InputSplit[] splits = getCombineSplits(job, numSplits, pathToPartitionInfo);
+ Collections.addAll(result, splits);
+ }
- // clear work from ThreadLocal after splits generated in case of thread is reused in pool.
- Utilities.clearWorkMapForConf(job);
+ // Restore the old path information back
+ // This is just to prevent incompatibilities with previous versions Hive
+ // if some application depends on the original value being set.
+ if (oldPaths != null) {
+ job.set(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR, oldPaths);
+ }
- // build internal schema for the query
- if (!result.isEmpty()) {
- ArrayList<String> uniqTablePaths = new ArrayList<>();
- Arrays.stream(paths).forEach(path -> {
- final HoodieStorage storage;
- try {
- FileSystem fs = path.getFileSystem(job);
- storage = HoodieStorageUtils.getStorage(
- HadoopFSUtils.convertToStoragePath(path), HadoopFSUtils.getStorageConf(fs.getConf()));
- Option<StoragePath> tablePath = TablePathUtils.getTablePath(storage, HadoopFSUtils.convertToStoragePath(path));
- if (tablePath.isPresent()) {
- uniqTablePaths.add(tablePath.get().toUri().toString());
+ // build internal schema for the query
+ if (!result.isEmpty()) {
+ ArrayList<String> uniqTablePaths = new ArrayList<>();
+ Arrays.stream(paths).forEach(path -> {
+ final HoodieStorage storage;
+ try {
+ FileSystem fs = path.getFileSystem(job);
+ storage = HoodieStorageUtils.getStorage(
+ HadoopFSUtils.convertToStoragePath(path), HadoopFSUtils.getStorageConf(fs.getConf()));
+ Option<StoragePath> tablePath = TablePathUtils.getTablePath(storage, HadoopFSUtils.convertToStoragePath(path));
+ if (tablePath.isPresent()) {
+ uniqTablePaths.add(tablePath.get().toUri().toString());
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
}
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- });
+ });
- try {
- for (String path : uniqTablePaths) {
- HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setBasePath(path).setConf(new HadoopStorageConfiguration(job)).build();
- TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
- String avroSchema = schemaUtil.getTableSchema().toString();
- Option<InternalSchema> internalSchema = schemaUtil.getTableInternalSchemaFromCommitMetadata();
- if (internalSchema.isPresent()) {
- LOG.info("Set internal and avro schema cache with path: {}", path);
- job.set(SCHEMA_CACHE_KEY_PREFIX + "." + path, avroSchema);
- job.set(INTERNAL_SCHEMA_CACHE_KEY_PREFIX + "." + path, SerDeHelper.toJson(internalSchema.get()));
- } else {
- // always sets up the cache so that we can distinguish with the scenario where the cache was never set(e.g. in tests).
- job.set(SCHEMA_CACHE_KEY_PREFIX + "." + path, "");
- job.set(INTERNAL_SCHEMA_CACHE_KEY_PREFIX + "." + path, "");
+ try {
+ for (String path : uniqTablePaths) {
+ HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder().setBasePath(path).setConf(new HadoopStorageConfiguration(job)).build();
+ TableSchemaResolver schemaUtil = new TableSchemaResolver(metaClient);
+ String avroSchema = schemaUtil.getTableSchema().toString();
+ Option<InternalSchema> internalSchema = schemaUtil.getTableInternalSchemaFromCommitMetadata();
+ if (internalSchema.isPresent()) {
+ LOG.info("Set internal and avro schema cache with path: {}", path);
+ job.set(SCHEMA_CACHE_KEY_PREFIX + "." + path, avroSchema);
+ job.set(INTERNAL_SCHEMA_CACHE_KEY_PREFIX + "." + path, SerDeHelper.toJson(internalSchema.get()));
+ } else {
+ // always sets up the cache so that we can distinguish with the scenario where the cache was never set(e.g. in tests).
+ job.set(SCHEMA_CACHE_KEY_PREFIX + "." + path, "");
+ job.set(INTERNAL_SCHEMA_CACHE_KEY_PREFIX + "." + path, "");
+ }
}
+ } catch (Exception e) {
+ LOG.warn("Failed to set schema cache", e);
}
- } catch (Exception e) {
- LOG.warn("Failed to set schema cache", e);
}
- }
- LOG.info("Number of all splits {}", result.size());
- perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
- return result.toArray(new InputSplit[result.size()]);
+ LOG.info("Number of all splits {}", result.size());
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
+ return result.toArray(new InputSplit[result.size()]);
+ } finally {
+ // Clear work from ThreadLocal after each getSplits attempt, in case the thread is reused in a pool.
+ Utilities.clearWorkMapForConf(job);
+ }
}
private void processPaths(JobConf job, CombineFileInputFormatShim combine, List<CombineFileSplit> iss, Path... path)
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
index bf331eb55c84..44ceac534004 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/hive/TestHoodieCombineHiveInputFormat.java
@@ -67,6 +67,8 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
@@ -80,7 +82,13 @@ import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_MAP_WORK;
import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_MAPPER_CLASS;
import static org.apache.hudi.common.testutils.HoodieTestUtils.COMMIT_METADATA_SER_DE;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
public class TestHoodieCombineHiveInputFormat extends HoodieCommonTestHarness {
@@ -119,6 +127,30 @@ public class TestHoodieCombineHiveInputFormat extends HoodieCommonTestHarness {
}
}
+ @Test
+ public void testClearWorkMapForConfOnGetSplitsFailure() throws Exception {
+ StorageConfiguration<Configuration> conf = HoodieTestUtils.getDefaultStorageConf();
+ File inputDir = tempDir.resolve("input").toFile();
+ assertTrue(inputDir.mkdirs());
+
+ MapredWork mrwork = new MapredWork();
+ Path mapWorkPath = new Path(tempDir.toAbsolutePath().toString());
+ Utilities.setMapRedWork(conf.unwrap(), mrwork, mapWorkPath);
+ JobConf jobConf = new JobConf(conf.unwrap());
+ FileInputFormat.setInputPaths(jobConf, inputDir.getPath());
+ jobConf.set(HAS_MAP_WORK, "true");
+ jobConf.set(MAPRED_MAPPER_CLASS, ExecMapper.class.getName());
+
+ HoodieCombineHiveInputFormat combineHiveInputFormat = spy(new HoodieCombineHiveInputFormat());
+ doThrow(new RuntimeException("path classification failed")).when(combineHiveInputFormat)
+ .getNonCombinablePathIndices(eq(jobConf), any(Path[].class), anyInt());
+
+ try (MockedStatic<Utilities> utilities = Mockito.mockStatic(Utilities.class, Mockito.CALLS_REAL_METHODS)) {
+ assertThrows(IOException.class, () -> combineHiveInputFormat.getSplits(jobConf, 1));
+ utilities.verify(() -> Utilities.clearWorkMapForConf(jobConf));
+ }
+ }
+
@Test
public void testInternalSchemaCacheForMR() throws Exception {
// test for HUDI-8182