HBASE-30501 Preserve original HFiles when splitting for bulk load

hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java

@@ -718,7 +718,9 @@ private List<LoadQueueItem> splitStoreFile(final LoadQueueItem item, final Table
     // we don't need it anymore. Clean up to save space.
     // It is not part of the original input files.
     try {
-      if (tmpDir.getName().equals(TMP_DIR)) {
+      // Preserve original input files so a new bulk load invocation can retry if loading the split
+      // outputs fails. Only delete an input which was itself produced by an earlier split.
+      if (hfilePath.getParent().getName().equals(TMP_DIR)) {
         fs.delete(hfilePath, false);
       }
     } catch (IOException e) {

hbase-server/src/test/java/org/apache/hadoop/hbase/tool/BulkLoadHFilesSplitRecoveryTestBase.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
@@ -473,9 +474,74 @@ protected void bulkLoadPhase(Table table, Connection conn, ExecutorService pool,
     assertEquals(4, bulkloadRpcTimes.get());
   }
 
+  @Test
+  public void testFreshRetryAfterSplitFailure(TestInfo testInfo) throws Exception {
+    TableName table = TableName.valueOf(testInfo.getTestMethod().get().getName());
+    setupTableWithSplitkeys(table, NUM_CFS, new byte[][] { rowkey(ROWCOUNT / 2) });
+
+    Path dir = buildBulkFiles(table, 1);
+    FileSystem fs = dir.getFileSystem(util.getConfiguration());
+    Path familyDir = new Path(dir, family(0));
+    Path originalHFile = new Path(familyDir, "hfile_0");
+    Path tmpDir = new Path(familyDir, LoadIncrementalHFiles.TMP_DIR);
+    assertTrue(fs.exists(originalHFile));
+
+    LoadIncrementalHFiles failingLoader = new LoadIncrementalHFiles(util.getConfiguration()) {
+      @Override
+      protected void bulkLoadPhase(Table table, Connection conn, ExecutorService pool,
+        Deque<LoadQueueItem> queue, Multimap<ByteBuffer, LoadQueueItem> regionGroups,
+        boolean copyFile, Map<LoadQueueItem, ByteBuffer> item2RegionMap) throws IOException {
+        throw new IOException("failure after splitting");
+      }
+    };
+
+    try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration());
+      Table t = connection.getTable(table);
+      RegionLocator locator = connection.getRegionLocator(table);
+      Admin admin = connection.getAdmin()) {
+      IOException error =
+        assertThrows(IOException.class, () -> failingLoader.doBulkLoad(dir, admin, t, locator));
+      assertEquals("failure after splitting", error.getMessage());
+    }
+
+    assertTrue(fs.exists(originalHFile));
+    FileStatus[] abandonedChildren = CommonFSUtils.listStatus(fs, tmpDir);
+    assertNotNull(abandonedChildren);
+    assertEquals(2, abandonedChildren.length);
+    Path top = null;
+    Path bottom = null;
+    for (FileStatus child : abandonedChildren) {
+      if (child.getPath().getName().endsWith(".top")) {
+        top = child.getPath();
+      } else if (child.getPath().getName().endsWith(".bottom")) {
+        bottom = child.getPath();
+      }
+    }
+    assertNotNull(top);
+    assertNotNull(bottom);
+
+    // Use a new loader to simulate retrying in another process. The original HFiles must remain in
+    // the family directory so the retry does not depend on in-memory references to split outputs.
+    try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration());
+      Table t = connection.getTable(table);
+      RegionLocator locator = connection.getRegionLocator(table);
+      Admin admin = connection.getAdmin()) {
+      new LoadIncrementalHFiles(util.getConfiguration()).doBulkLoad(dir, admin, t, locator);
+    }
+
+    // The retry consumes the preserved original HFiles. Split outputs abandoned by the failed
+    // invocation remain temporary and are deliberately not treated as input because they may be
+    // incomplete.
+    assertTrue(fs.exists(originalHFile));
+    assertTrue(fs.exists(top));
+    assertTrue(fs.exists(bottom));
+    assertEquals(2, CommonFSUtils.listStatus(fs, tmpDir).length);
+    assertExpectedTable(table, ROWCOUNT, 1);
+  }
+
   /**
-   * This test creates a table with many small regions. The bulk load files would be splitted
-   * multiple times before all of them can be loaded successfully.
+   * This test creates a table with many small regions. The bulk load files would be split multiple
+   * times before all of them can be loaded successfully.
    */
   @Test
   public void testSplitTmpFileCleanUp(TestInfo testInfo) throws Exception {