1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import static java.util.Arrays.asList;
22 import static org.junit.Assert.assertEquals;
23 import static org.junit.Assert.assertNotNull;
24 import static org.junit.Assert.assertTrue;
25
26 import java.io.File;
27 import java.io.FileNotFoundException;
28 import java.io.FileOutputStream;
29 import java.io.IOException;
30 import java.util.ArrayList;
31 import java.util.Arrays;
32 import java.util.List;
33 import java.util.Random;
34
35 import org.apache.hadoop.conf.Configuration;
36 import org.apache.hadoop.fs.FSDataOutputStream;
37 import org.apache.hadoop.fs.Path;
38 import org.apache.hadoop.hbase.CellUtil;
39 import org.apache.hadoop.hbase.DoNotRetryIOException;
40 import org.apache.hadoop.hbase.HBaseConfiguration;
41 import org.apache.hadoop.hbase.HColumnDescriptor;
42 import org.apache.hadoop.hbase.HRegionInfo;
43 import org.apache.hadoop.hbase.HTableDescriptor;
44 import org.apache.hadoop.hbase.KeyValue;
45 import org.apache.hadoop.hbase.TableName;
46 import org.apache.hadoop.hbase.io.hfile.HFile;
47 import org.apache.hadoop.hbase.io.hfile.HFileContext;
48 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
49 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.BulkLoadDescriptor;
50 import org.apache.hadoop.hbase.protobuf.generated.WALProtos.StoreDescriptor;
51 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
52 import org.apache.hadoop.hbase.testclassification.SmallTests;
53 import org.apache.hadoop.hbase.util.Bytes;
54 import org.apache.hadoop.hbase.util.Pair;
55 import org.apache.hadoop.hbase.wal.WAL;
56 import org.apache.hadoop.hbase.wal.WALKey;
57 import org.hamcrest.Description;
58 import org.hamcrest.Matcher;
59 import org.hamcrest.TypeSafeMatcher;
60
61 import static org.mockito.Matchers.*;
62 import static org.mockito.Mockito.mock;
63 import static org.mockito.Mockito.verify;
64 import static org.mockito.Mockito.when;
65 import org.mockito.invocation.InvocationOnMock;
66 import org.mockito.stubbing.Answer;
67
68 import org.junit.Before;
69 import org.junit.ClassRule;
70 import org.junit.Rule;
71 import org.junit.Test;
72 import org.junit.experimental.categories.Category;
73 import org.junit.rules.TemporaryFolder;
74 import org.junit.rules.TestName;
75
76
77
78
79 @Category(SmallTests.class)
80 public class TestBulkLoad {
81
82 @ClassRule
83 public static TemporaryFolder testFolder = new TemporaryFolder();
84 private final WAL log = mock(WAL.class);
85 private final Configuration conf = HBaseConfiguration.create();
86 private final Random random = new Random();
87 private final byte[] randomBytes = new byte[100];
88 private final byte[] family1 = Bytes.toBytes("family1");
89 private final byte[] family2 = Bytes.toBytes("family2");
90 @Rule
91 public TestName name = new TestName();
92
93 @Before
94 public void before() throws IOException {
95 random.nextBytes(randomBytes);
96
97 }
98
99 @Test
100 public void verifyBulkLoadEvent() throws IOException {
101 TableName tableName = TableName.valueOf("test", "test");
102 List<Pair<byte[], String>> familyPaths = withFamilyPathsFor(family1);
103 byte[] familyName = familyPaths.get(0).getFirst();
104 String storeFileName = familyPaths.get(0).getSecond();
105 storeFileName = (new Path(storeFileName)).getName();
106 List<String> storeFileNames = new ArrayList<String>();
107 storeFileNames.add(storeFileName);
108 when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class), any(WALKey.class),
109 argThat(bulkLogWalEdit(WALEdit.BULK_LOAD, tableName.toBytes(),
110 familyName, storeFileNames)),
111 any(boolean.class))).thenAnswer(new Answer() {
112 public Object answer(InvocationOnMock invocation) {
113 WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
114 MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
115 if (mvcc != null) {
116 MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
117 walKey.setWriteEntry(we);
118 }
119 return 01L;
120 };
121 });
122 testRegionWithFamiliesAndSpecifiedTableName(tableName, family1)
123 .bulkLoadHFiles(familyPaths, false, null);
124 verify(log).sync(anyLong());
125 }
126
127 @Test
128 public void bulkHLogShouldThrowNoErrorAndWriteMarkerWithBlankInput() throws IOException {
129 testRegionWithFamilies(family1).bulkLoadHFiles(new ArrayList<Pair<byte[], String>>(),
130 false, null);
131 }
132
133 @Test
134 public void shouldBulkLoadSingleFamilyHLog() throws IOException {
135 when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class),
136 any(WALKey.class), argThat(bulkLogWalEditType(WALEdit.BULK_LOAD)),
137 any(boolean.class))).thenAnswer(new Answer() {
138 public Object answer(InvocationOnMock invocation) {
139 WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
140 MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
141 if (mvcc != null) {
142 MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
143 walKey.setWriteEntry(we);
144 }
145 return 01L;
146 };
147 });
148 testRegionWithFamilies(family1).bulkLoadHFiles(withFamilyPathsFor(family1), false, null);
149 verify(log).sync(anyLong());
150 }
151
152 @Test
153 public void shouldBulkLoadManyFamilyHLog() throws IOException {
154 when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class),
155 any(WALKey.class), argThat(bulkLogWalEditType(WALEdit.BULK_LOAD)),
156 any(boolean.class))).thenAnswer(new Answer() {
157 public Object answer(InvocationOnMock invocation) {
158 WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
159 MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
160 if (mvcc != null) {
161 MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
162 walKey.setWriteEntry(we);
163 }
164 return 01L;
165 };
166 });
167 testRegionWithFamilies(family1, family2).bulkLoadHFiles(withFamilyPathsFor(family1, family2),
168 false, null);
169 verify(log).sync(anyLong());
170 }
171
172 @Test
173 public void shouldBulkLoadManyFamilyHLogEvenWhenTableNameNamespaceSpecified() throws IOException {
174 when(log.append(any(HTableDescriptor.class), any(HRegionInfo.class),
175 any(WALKey.class), argThat(bulkLogWalEditType(WALEdit.BULK_LOAD)),
176 any(boolean.class))).thenAnswer(new Answer() {
177 public Object answer(InvocationOnMock invocation) {
178 WALKey walKey = invocation.getArgumentAt(2, WALKey.class);
179 MultiVersionConcurrencyControl mvcc = walKey.getMvcc();
180 if (mvcc != null) {
181 MultiVersionConcurrencyControl.WriteEntry we = mvcc.begin();
182 walKey.setWriteEntry(we);
183 }
184 return 01L;
185 };
186 });
187 TableName tableName = TableName.valueOf("test", "test");
188 testRegionWithFamiliesAndSpecifiedTableName(tableName, family1, family2)
189 .bulkLoadHFiles(withFamilyPathsFor(family1, family2), false, null);
190 verify(log).sync(anyLong());
191 }
192
193 @Test(expected = DoNotRetryIOException.class)
194 public void shouldCrashIfBulkLoadFamiliesNotInTable() throws IOException {
195 testRegionWithFamilies(family1).bulkLoadHFiles(withFamilyPathsFor(family1, family2), false,
196 null);
197 }
198
199 @Test(expected = DoNotRetryIOException.class)
200 public void bulkHLogShouldThrowErrorWhenFamilySpecifiedAndHFileExistsButNotInTableDescriptor()
201 throws IOException {
202 testRegionWithFamilies().bulkLoadHFiles(withFamilyPathsFor(family1), false, null);
203 }
204
205 @Test(expected = DoNotRetryIOException.class)
206 public void shouldThrowErrorIfBadFamilySpecifiedAsFamilyPath() throws IOException {
207 testRegionWithFamilies()
208 .bulkLoadHFiles(asList(withInvalidColumnFamilyButProperHFileLocation(family1)),
209 false, null);
210 }
211
212 @Test(expected = FileNotFoundException.class)
213 public void shouldThrowErrorIfHFileDoesNotExist() throws IOException {
214 List<Pair<byte[], String>> list = asList(withMissingHFileForFamily(family1));
215 testRegionWithFamilies(family1).bulkLoadHFiles(list, false, null);
216 }
217
218 private Pair<byte[], String> withMissingHFileForFamily(byte[] family) {
219 return new Pair<byte[], String>(family, "/tmp/does_not_exist");
220 }
221
222 private Pair<byte[], String> withInvalidColumnFamilyButProperHFileLocation(byte[] family)
223 throws IOException {
224 createHFileForFamilies(family);
225 return new Pair<byte[], String>(new byte[]{0x00, 0x01, 0x02}, "/tmp/does_not_exist");
226 }
227
228
229 private HRegion testRegionWithFamiliesAndSpecifiedTableName(TableName tableName,
230 byte[]... families)
231 throws IOException {
232 HRegionInfo hRegionInfo = new HRegionInfo(tableName);
233 HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
234 for (byte[] family : families) {
235 hTableDescriptor.addFamily(new HColumnDescriptor(family));
236 }
237
238
239 return HRegion.createHRegion(hRegionInfo,
240 new Path(testFolder.newFolder().toURI()),
241 conf,
242 hTableDescriptor,
243 log);
244
245 }
246
247 private HRegion testRegionWithFamilies(byte[]... families) throws IOException {
248 TableName tableName = TableName.valueOf(name.getMethodName());
249 return testRegionWithFamiliesAndSpecifiedTableName(tableName, families);
250 }
251
252 private List<Pair<byte[], String>> getBlankFamilyPaths(){
253 return new ArrayList<Pair<byte[], String>>();
254 }
255
256 private List<Pair<byte[], String>> withFamilyPathsFor(byte[]... families) throws IOException {
257 List<Pair<byte[], String>> familyPaths = getBlankFamilyPaths();
258 for (byte[] family : families) {
259 familyPaths.add(new Pair<byte[], String>(family, createHFileForFamilies(family)));
260 }
261 return familyPaths;
262 }
263
264 private String createHFileForFamilies(byte[] family) throws IOException {
265 HFile.WriterFactory hFileFactory = HFile.getWriterFactoryNoCache(conf);
266
267 File hFileLocation = testFolder.newFile();
268 FSDataOutputStream out = new FSDataOutputStream(new FileOutputStream(hFileLocation));
269 try {
270 hFileFactory.withOutputStream(out);
271 hFileFactory.withFileContext(new HFileContext());
272 HFile.Writer writer = hFileFactory.create();
273 try {
274 writer.append(new KeyValue(CellUtil.createCell(randomBytes,
275 family,
276 randomBytes,
277 0l,
278 KeyValue.Type.Put.getCode(),
279 randomBytes)));
280 } finally {
281 writer.close();
282 }
283 } finally {
284 out.close();
285 }
286 return hFileLocation.getAbsoluteFile().getAbsolutePath();
287 }
288
289 private static Matcher<WALEdit> bulkLogWalEditType(byte[] typeBytes) {
290 return new WalMatcher(typeBytes);
291 }
292
293 private static Matcher<WALEdit> bulkLogWalEdit(byte[] typeBytes, byte[] tableName,
294 byte[] familyName, List<String> storeFileNames) {
295 return new WalMatcher(typeBytes, tableName, familyName, storeFileNames);
296 }
297
298 private static class WalMatcher extends TypeSafeMatcher<WALEdit> {
299 private final byte[] typeBytes;
300 private final byte[] tableName;
301 private final byte[] familyName;
302 private final List<String> storeFileNames;
303
304 public WalMatcher(byte[] typeBytes) {
305 this(typeBytes, null, null, null);
306 }
307
308 public WalMatcher(byte[] typeBytes, byte[] tableName, byte[] familyName,
309 List<String> storeFileNames) {
310 this.typeBytes = typeBytes;
311 this.tableName = tableName;
312 this.familyName = familyName;
313 this.storeFileNames = storeFileNames;
314 }
315
316 @Override
317 protected boolean matchesSafely(WALEdit item) {
318 assertTrue(Arrays.equals(item.getCells().get(0).getQualifier(), typeBytes));
319 BulkLoadDescriptor desc;
320 try {
321 desc = WALEdit.getBulkLoadDescriptor(item.getCells().get(0));
322 } catch (IOException e) {
323 return false;
324 }
325 assertNotNull(desc);
326
327 if (tableName != null) {
328 assertTrue(Bytes.equals(ProtobufUtil.toTableName(desc.getTableName()).getName(),
329 tableName));
330 }
331
332 if(storeFileNames != null) {
333 int index=0;
334 StoreDescriptor store = desc.getStores(0);
335 assertTrue(Bytes.equals(store.getFamilyName().toByteArray(), familyName));
336 assertTrue(Bytes.equals(Bytes.toBytes(store.getStoreHomeDir()), familyName));
337 assertEquals(storeFileNames.size(), store.getStoreFileCount());
338 }
339
340 return true;
341 }
342
343 @Override
344 public void describeTo(Description description) {
345
346 }
347 }
348 }