From 1d33fea7601d5cf32780a8612ab95a5af31e5a00 Mon Sep 17 00:00:00 2001 From: mazhengxuan Date: Thu, 20 Aug 2026 23:30:34 +0800 Subject: [PATCH] HBASE-30288 Fix WALPlayer bulk output table mappings Signed-off-by: mazhengxuan --- .../hadoop/hbase/mapreduce/WALPlayer.java | 27 ++++++++------ .../hadoop/hbase/mapreduce/TestWALPlayer.java | 35 +++++++++++-------- 2 files changed, 38 insertions(+), 24 deletions(-) diff --git a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALPlayer.java b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALPlayer.java index 9813118e2502..c8697879479e 100644 --- a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALPlayer.java +++ b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/WALPlayer.java @@ -21,11 +21,8 @@ import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.ArrayList; -import java.util.Collections; -import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.TreeMap; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; @@ -99,7 +96,7 @@ protected WALPlayer(final Configuration c) { * {@link CellSortReducer} */ static class WALKeyValueMapper extends Mapper, Cell> { - private Set tableSet = new HashSet(); + private final Map tables = new TreeMap<>(); private boolean multiTableSupport = false; private boolean diskBasedSortingEnabled = false; @@ -107,8 +104,8 @@ static class WALKeyValueMapper extends Mapper wrapKey(byte[] key, ExtendedCell cell) { @@ -349,7 +356,7 @@ public Job createSubmittableJob(String[] args) throws IOException { true); // the bulk HFile case - List tableNames = getTableNameList(tables); + List tableNames = getTableNameList(tableMap); job.setMapperClass(WALKeyValueMapper.class); if (diskBasedSortingEnabled) { diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALPlayer.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALPlayer.java index ce59ef1424bd..d256e7bb3136 100644 --- a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALPlayer.java +++ b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALPlayer.java @@ -137,16 +137,23 @@ public void testPlayingRecoveredEdit() throws Exception { /** * Tests that when you write multiple cells with the same timestamp they are properly sorted by * their sequenceId in WALPlayer/CellSortReducer so that the correct one wins when querying from - * the resulting bulkloaded HFiles. See HBASE-27649 + * the resulting bulkloaded HFiles. Also verifies that bulk output honors source-to-target table + * mappings. See HBASE-27649 and HBASE-30288. */ @Test - public void testWALPlayerBulkLoadWithOverriddenTimestamps(TestInfo testInfo) throws Exception { - final TableName tableName = TableName.valueOf(testInfo.getTestMethod().get().getName() + "1"); + public void testWALPlayerBulkLoadWithTableMappingAndOverriddenTimestamps(TestInfo testInfo) + throws Exception { + final TableName sourceTableName = + TableName.valueOf(testInfo.getTestMethod().get().getName() + "Source"); + final TableName targetTableName = + TableName.valueOf(testInfo.getTestMethod().get().getName() + "Target"); final byte[] family = Bytes.toBytes("family"); final byte[] column1 = Bytes.toBytes("c1"); final byte[] column2 = Bytes.toBytes("c2"); final byte[] row = Bytes.toBytes("row"); - final Table table = TEST_UTIL.createTable(tableName, family); + final Table sourceTable = TEST_UTIL.createTable(sourceTableName, family); + final Table targetTable = + TEST_UTIL.createTable(targetTableName, family, new byte[][] { Bytes.toBytes("row0") }); long now = EnvironmentEdgeManager.currentTime(); // put a row into the first table @@ -154,7 +161,7 @@ public void testWALPlayerBulkLoadWithOverriddenTimestamps(TestInfo testInfo) thr p.addColumn(family, column1, now, column1); p.addColumn(family, column2, now, column2); - table.put(p); + sourceTable.put(p); byte[] lastVal = null; @@ -163,7 +170,7 @@ public void testWALPlayerBulkLoadWithOverriddenTimestamps(TestInfo testInfo) thr p = new Put(row); p.addColumn(family, column1, now, lastVal); - table.put(p); + sourceTable.put(p); // wal rolling is necessary to trigger the bug. otherwise no sorting // needs to occur in the reducer because it's all sorted and coming from a single file. @@ -187,24 +194,24 @@ public void testWALPlayerBulkLoadWithOverriddenTimestamps(TestInfo testInfo) thr final byte[] finalLastVal = lastVal; runWithDiskBasedSortingDisabledAndEnabled(() -> { - assertEquals(0, ToolRunner.run(configuration, player, - new String[] { walInputDir, tableName.getNameAsString() })); + assertEquals(0, ToolRunner.run(configuration, player, new String[] { walInputDir, + sourceTableName.getNameAsString(), targetTableName.getNameAsString() })); Get g = new Get(row); - Result result = table.get(g); + Result result = sourceTable.get(g); byte[] value = CellUtil.cloneValue(result.getColumnLatestCell(family, column1)); assertThat(Bytes.toStringBinary(value), equalTo(Bytes.toStringBinary(finalLastVal))); - TEST_UTIL.truncateTable(tableName); + TEST_UTIL.truncateTable(targetTableName); g = new Get(row); - result = table.get(g); + result = targetTable.get(g); assertThat(result.listCells(), nullValue()); - BulkLoadHFiles.create(configuration).bulkLoad(tableName, - new Path(outPath, tableName.getNamespaceAsString() + "/" + tableName.getNameAsString())); + BulkLoadHFiles.create(configuration).bulkLoad(targetTableName, new Path(outPath, + targetTableName.getNamespaceAsString() + "/" + targetTableName.getNameAsString())); g = new Get(row); - result = table.get(g); + result = targetTable.get(g); value = CellUtil.cloneValue(result.getColumnLatestCell(family, column1)); assertThat(result.listCells(), notNullValue());