001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.backup.util;
019
020import static org.apache.hadoop.hbase.wal.WALEdit.METAFAMILY;
021import static org.junit.jupiter.api.Assertions.assertEquals;
022import static org.junit.jupiter.api.Assertions.assertTrue;
023
024import java.io.IOException;
025import java.util.Arrays;
026import java.util.Collections;
027import java.util.List;
028import org.apache.hadoop.fs.Path;
029import org.apache.hadoop.hbase.Cell;
030import org.apache.hadoop.hbase.CellBuilderFactory;
031import org.apache.hadoop.hbase.CellBuilderType;
032import org.apache.hadoop.hbase.TableName;
033import org.apache.hadoop.hbase.testclassification.SmallTests;
034import org.apache.hadoop.hbase.util.Bytes;
035import org.apache.hadoop.hbase.wal.WAL;
036import org.apache.hadoop.hbase.wal.WALEdit;
037import org.apache.hadoop.hbase.wal.WALKeyImpl;
038import org.junit.jupiter.api.Tag;
039import org.junit.jupiter.api.Test;
040
041import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
042
043import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
044import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
045
046/**
047 * Unit tests for {@link BulkLoadProcessor}.
048 * <p>
049 * These tests validate the extraction of bulk-loaded file paths from WAL entries under different
050 * scenarios, including:
051 * <ul>
052 * <li>Valid replicable bulk load entries</li>
053 * <li>Non-replicable bulk load entries</li>
054 * <li>Entries with no bulk load qualifier</li>
055 * <li>Entries containing multiple column families</li>
056 * </ul>
057 */
058@Tag(SmallTests.TAG)
059public class TestBulkLoadProcessor {
060
061  /**
062   * Creates a WAL.Entry containing a {@link WALProtos.BulkLoadDescriptor} with the given
063   * parameters.
064   * @param tableName  The table name
065   * @param regionName The encoded region name
066   * @param replicate  Whether the bulk load is marked for replication
067   * @param family     Column family name
068   * @param storeFiles One or more store file names to include
069   * @return A WAL.Entry representing the bulk load event
070   */
071  private WAL.Entry createBulkLoadWalEntry(TableName tableName, String regionName,
072    boolean replicate, String family, String... storeFiles) {
073
074    // Build StoreDescriptor
075    WALProtos.StoreDescriptor.Builder storeDescBuilder =
076      WALProtos.StoreDescriptor.newBuilder().setFamilyName(ByteString.copyFromUtf8(family))
077        .setStoreHomeDir(family).addAllStoreFile(Arrays.asList(storeFiles));
078
079    // Build BulkLoadDescriptor
080    WALProtos.BulkLoadDescriptor.Builder bulkDescBuilder = WALProtos.BulkLoadDescriptor.newBuilder()
081      .setReplicate(replicate).setEncodedRegionName(ByteString.copyFromUtf8(regionName))
082      .setTableName(ProtobufUtil.toProtoTableName(tableName)).setBulkloadSeqNum(1000) // Random
083      .addStores(storeDescBuilder);
084
085    byte[] value = bulkDescBuilder.build().toByteArray();
086
087    // Build Cell with BULK_LOAD qualifier
088    Cell cell = CellBuilderFactory.create(CellBuilderType.DEEP_COPY).setType(Cell.Type.Put)
089      .setRow(new byte[] { 1 }).setFamily(METAFAMILY).setQualifier(WALEdit.BULK_LOAD)
090      .setValue(value).build();
091
092    WALEdit edit = new WALEdit();
093    edit.add(cell);
094
095    WALKeyImpl key = new WALKeyImpl(Bytes.toBytes(regionName), // region
096      tableName, 0L, 0L, null);
097
098    return new WAL.Entry(key, edit);
099  }
100
101  /**
102   * Verifies that a valid replicable bulk load WAL entry produces the correct number and structure
103   * of file paths.
104   */
105  @Test
106  public void testProcessBulkLoadFiles_validEntry() throws IOException {
107    WAL.Entry entry = createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", true,
108      "cf1", "file1", "file2");
109
110    List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry));
111
112    assertEquals(2, paths.size());
113    assertTrue(paths.get(0).toString().contains("ns/tbl/region123/cf1/file1"));
114    assertTrue(paths.get(1).toString().contains("ns/tbl/region123/cf1/file2"));
115  }
116
117  @Test
118  public void testProcessBulkLoadFiles_validEntry_singleEntryApi() throws IOException {
119    WAL.Entry entry = createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", true,
120      "cf1", "file1", "file2");
121
122    List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit());
123
124    assertEquals(2, paths.size());
125    assertTrue(paths.get(0).toString().contains("ns/tbl/region123/cf1/file1"));
126    assertTrue(paths.get(1).toString().contains("ns/tbl/region123/cf1/file2"));
127  }
128
129  /**
130   * Verifies that a non-replicable bulk load entry is ignored.
131   */
132  @Test
133  public void testProcessBulkLoadFiles_nonReplicableSkipped() throws IOException {
134    WAL.Entry entry =
135      createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", false, "cf1", "file1");
136
137    List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry));
138
139    assertTrue(paths.isEmpty());
140  }
141
142  @Test
143  public void testProcessBulkLoadFiles_nonReplicableSkipped_singleEntryApi() throws IOException {
144    WAL.Entry entry =
145      createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", false, "cf1", "file1");
146
147    List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit());
148
149    assertTrue(paths.isEmpty());
150  }
151
152  /**
153   * Verifies that entries without the BULK_LOAD qualifier are ignored.
154   */
155  @Test
156  public void testProcessBulkLoadFiles_noBulkLoadQualifier() throws IOException {
157    WALEdit edit = new WALEdit();
158    WALKeyImpl key = new WALKeyImpl(new byte[] {}, TableName.valueOf("ns", "tbl"), 0L, 0L, null);
159    WAL.Entry entry = new WAL.Entry(key, edit);
160
161    List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry));
162
163    assertTrue(paths.isEmpty());
164  }
165
166  @Test
167  public void testProcessBulkLoadFiles_noBulkLoadQualifier_singleEntryApi() throws IOException {
168    WALEdit edit = new WALEdit();
169    WALKeyImpl key = new WALKeyImpl(new byte[] {}, TableName.valueOf("ns", "tbl"), 0L, 0L, null);
170
171    List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(key, edit);
172
173    assertTrue(paths.isEmpty());
174  }
175
176  /**
177   * Verifies that multiple WAL entries with different column families produce the correct set of
178   * file paths.
179   */
180  @Test
181  public void testProcessBulkLoadFiles_multipleFamilies() throws IOException {
182    WAL.Entry entry =
183      createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf1", "file1");
184    WAL.Entry entry2 =
185      createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf2", "fileA");
186
187    List<Path> paths = BulkLoadProcessor.processBulkLoadFiles(Arrays.asList(entry, entry2));
188
189    assertEquals(2, paths.size());
190    assertTrue(paths.stream().anyMatch(p -> p.toString().contains("cf1/file1")));
191    assertTrue(paths.stream().anyMatch(p -> p.toString().contains("cf2/fileA")));
192  }
193
194  @Test
195  public void testProcessBulkLoadFiles_multipleFamilies_singleEntryApi() throws IOException {
196    WAL.Entry entry =
197      createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf1", "file1");
198    WAL.Entry entry2 =
199      createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "regionXYZ", true, "cf2", "fileA");
200
201    List<Path> paths1 = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit());
202    List<Path> paths2 = BulkLoadProcessor.processBulkLoadFiles(entry2.getKey(), entry2.getEdit());
203
204    // combine to mimic processing multiple entries
205    paths1.addAll(paths2);
206
207    assertEquals(2, paths1.size());
208    assertTrue(paths1.stream().anyMatch(p -> p.toString().contains("cf1/file1")));
209    assertTrue(paths1.stream().anyMatch(p -> p.toString().contains("cf2/fileA")));
210  }
211
212  /**
213   * Sanity check: list-based API should still work and return the same results as invoking the
214   * single-entry API for the same entry (ensures delegation/backwards compatibility).
215   */
216  @Test
217  public void testProcessBulkLoadFiles_listApi_delegatesToSingle() throws IOException {
218    WAL.Entry entry =
219      createBulkLoadWalEntry(TableName.valueOf("ns", "tbl"), "region123", true, "cf1", "file1");
220
221    List<Path> single = BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit());
222    List<Path> listApi = BulkLoadProcessor.processBulkLoadFiles(Collections.singletonList(entry));
223
224    assertEquals(single.size(), listApi.size());
225    assertTrue(listApi.get(0).toString().contains("ns/tbl/region123/cf1/file1"));
226  }
227}