1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mapreduce;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertNotNull;
24 import static org.junit.Assert.assertNotSame;
25 import static org.junit.Assert.assertTrue;
26 import static org.junit.Assert.fail;
27
28 import java.io.IOException;
29 import java.util.Arrays;
30 import java.util.HashMap;
31 import java.util.List;
32 import java.util.Map;
33 import java.util.Map.Entry;
34 import java.util.Random;
35 import java.util.Set;
36 import java.util.concurrent.Callable;
37
38 import junit.framework.Assert;
39
40 import org.apache.commons.logging.Log;
41 import org.apache.commons.logging.LogFactory;
42 import org.apache.hadoop.conf.Configuration;
43 import org.apache.hadoop.fs.FileStatus;
44 import org.apache.hadoop.fs.FileSystem;
45 import org.apache.hadoop.fs.Path;
46 import org.apache.hadoop.hbase.CategoryBasedTimeout;
47 import org.apache.hadoop.hbase.Cell;
48 import org.apache.hadoop.hbase.CellUtil;
49 import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
50 import org.apache.hadoop.hbase.HBaseConfiguration;
51 import org.apache.hadoop.hbase.HBaseTestingUtility;
52 import org.apache.hadoop.hbase.HColumnDescriptor;
53 import org.apache.hadoop.hbase.HConstants;
54 import org.apache.hadoop.hbase.HTableDescriptor;
55 import org.apache.hadoop.hbase.HadoopShims;
56 import org.apache.hadoop.hbase.KeyValue;
57 import org.apache.hadoop.hbase.PerformanceEvaluation;
58 import org.apache.hadoop.hbase.TableName;
59 import org.apache.hadoop.hbase.client.HBaseAdmin;
60 import org.apache.hadoop.hbase.client.HTable;
61 import org.apache.hadoop.hbase.client.Put;
62 import org.apache.hadoop.hbase.client.RegionLocator;
63 import org.apache.hadoop.hbase.client.Result;
64 import org.apache.hadoop.hbase.client.ResultScanner;
65 import org.apache.hadoop.hbase.client.Scan;
66 import org.apache.hadoop.hbase.client.Table;
67 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
68 import org.apache.hadoop.hbase.io.compress.Compression;
69 import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
70 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
71 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
72 import org.apache.hadoop.hbase.io.hfile.HFile;
73 import org.apache.hadoop.hbase.io.hfile.HFile.Reader;
74 import org.apache.hadoop.hbase.regionserver.BloomType;
75 import org.apache.hadoop.hbase.regionserver.HRegion;
76 import org.apache.hadoop.hbase.regionserver.HStore;
77 import org.apache.hadoop.hbase.regionserver.Store;
78 import org.apache.hadoop.hbase.regionserver.StoreFile;
79 import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
80 import org.apache.hadoop.hbase.testclassification.LargeTests;
81 import org.apache.hadoop.hbase.util.Bytes;
82 import org.apache.hadoop.hbase.util.FSUtils;
83 import org.apache.hadoop.hbase.util.Threads;
84 import org.apache.hadoop.hbase.util.Writables;
85 import org.apache.hadoop.io.NullWritable;
86 import org.apache.hadoop.mapreduce.Job;
87 import org.apache.hadoop.mapreduce.Mapper;
88 import org.apache.hadoop.mapreduce.RecordWriter;
89 import org.apache.hadoop.mapreduce.TaskAttemptContext;
90 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
91 import org.junit.Ignore;
92 import org.junit.Rule;
93 import org.junit.Test;
94 import org.junit.experimental.categories.Category;
95 import org.junit.rules.TestRule;
96 import org.mockito.Mockito;
97
98
99
100
101
102
103
104 @Category(LargeTests.class)
105 public class TestHFileOutputFormat {
106 @Rule public final TestRule timeout = CategoryBasedTimeout.builder().
107 withTimeout(this.getClass()).withLookingForStuckThread(true).build();
108 private final static int ROWSPERSPLIT = 1024;
109
110 private static final byte[][] FAMILIES
111 = { Bytes.add(PerformanceEvaluation.FAMILY_NAME, Bytes.toBytes("-A"))
112 , Bytes.add(PerformanceEvaluation.FAMILY_NAME, Bytes.toBytes("-B"))};
113 private static final TableName TABLE_NAME =
114 TableName.valueOf("TestTable");
115
116 private HBaseTestingUtility util = new HBaseTestingUtility();
117
118 private static final Log LOG = LogFactory.getLog(TestHFileOutputFormat.class);
119
120
121
122
123 static class RandomKVGeneratingMapper extends
124 Mapper<NullWritable, NullWritable, ImmutableBytesWritable, KeyValue> {
125
126 private int keyLength;
127 private static final int KEYLEN_DEFAULT = 10;
128 private static final String KEYLEN_CONF = "randomkv.key.length";
129
130 private int valLength;
131 private static final int VALLEN_DEFAULT=10;
132 private static final String VALLEN_CONF="randomkv.val.length";
133 private static final byte [] QUALIFIER = Bytes.toBytes("data");
134
135 @Override
136 protected void setup(Context context) throws IOException,
137 InterruptedException {
138 super.setup(context);
139
140 Configuration conf = context.getConfiguration();
141 keyLength = conf.getInt(KEYLEN_CONF, KEYLEN_DEFAULT);
142 valLength = conf.getInt(VALLEN_CONF, VALLEN_DEFAULT);
143 }
144
145 protected void map(
146 NullWritable n1, NullWritable n2,
147 Mapper<NullWritable, NullWritable,
148 ImmutableBytesWritable,KeyValue>.Context context)
149 throws java.io.IOException ,InterruptedException
150 {
151
152 byte keyBytes[] = new byte[keyLength];
153 byte valBytes[] = new byte[valLength];
154
155 int taskId = context.getTaskAttemptID().getTaskID().getId();
156 assert taskId < Byte.MAX_VALUE : "Unit tests dont support > 127 tasks!";
157
158 Random random = new Random();
159 for (int i = 0; i < ROWSPERSPLIT; i++) {
160
161 random.nextBytes(keyBytes);
162
163 keyBytes[keyLength - 1] = (byte)(taskId & 0xFF);
164 random.nextBytes(valBytes);
165 ImmutableBytesWritable key = new ImmutableBytesWritable(keyBytes);
166
167 for (byte[] family : TestHFileOutputFormat.FAMILIES) {
168 KeyValue kv = new KeyValue(keyBytes, family, QUALIFIER, valBytes);
169 context.write(key, kv);
170 }
171 }
172 }
173 }
174
175 private void setupRandomGeneratorMapper(Job job) {
176 job.setInputFormatClass(NMapInputFormat.class);
177 job.setMapperClass(RandomKVGeneratingMapper.class);
178 job.setMapOutputKeyClass(ImmutableBytesWritable.class);
179 job.setMapOutputValueClass(KeyValue.class);
180 }
181
182
183
184
185
186
187 @Test
188 public void test_LATEST_TIMESTAMP_isReplaced()
189 throws Exception {
190 Configuration conf = new Configuration(this.util.getConfiguration());
191 RecordWriter<ImmutableBytesWritable, KeyValue> writer = null;
192 TaskAttemptContext context = null;
193 Path dir =
194 util.getDataTestDir("test_LATEST_TIMESTAMP_isReplaced");
195 try {
196 Job job = new Job(conf);
197 FileOutputFormat.setOutputPath(job, dir);
198 context = createTestTaskAttemptContext(job);
199 HFileOutputFormat hof = new HFileOutputFormat();
200 writer = hof.getRecordWriter(context);
201 final byte [] b = Bytes.toBytes("b");
202
203
204
205 KeyValue kv = new KeyValue(b, b, b);
206 KeyValue original = kv.clone();
207 writer.write(new ImmutableBytesWritable(), kv);
208 assertFalse(original.equals(kv));
209 assertTrue(Bytes.equals(original.getRow(), kv.getRow()));
210 assertTrue(CellUtil.matchingColumn(original, kv.getFamily(), kv.getQualifier()));
211 assertNotSame(original.getTimestamp(), kv.getTimestamp());
212 assertNotSame(HConstants.LATEST_TIMESTAMP, kv.getTimestamp());
213
214
215
216 kv = new KeyValue(b, b, b, kv.getTimestamp() - 1, b);
217 original = kv.clone();
218 writer.write(new ImmutableBytesWritable(), kv);
219 assertTrue(original.equals(kv));
220 } finally {
221 if (writer != null && context != null) writer.close(context);
222 dir.getFileSystem(conf).delete(dir, true);
223 }
224 }
225
226 private TaskAttemptContext createTestTaskAttemptContext(final Job job)
227 throws IOException, Exception {
228 HadoopShims hadoop = CompatibilitySingletonFactory.getInstance(HadoopShims.class);
229 TaskAttemptContext context = hadoop.createTestTaskAttemptContext(job, "attempt_200707121733_0001_m_000000_0");
230 return context;
231 }
232
233
234
235
236
237 @Test
238 public void test_TIMERANGE() throws Exception {
239 Configuration conf = new Configuration(this.util.getConfiguration());
240 RecordWriter<ImmutableBytesWritable, KeyValue> writer = null;
241 TaskAttemptContext context = null;
242 Path dir =
243 util.getDataTestDir("test_TIMERANGE_present");
244 LOG.info("Timerange dir writing to dir: "+ dir);
245 try {
246
247 Job job = new Job(conf);
248 FileOutputFormat.setOutputPath(job, dir);
249 context = createTestTaskAttemptContext(job);
250 HFileOutputFormat hof = new HFileOutputFormat();
251 writer = hof.getRecordWriter(context);
252
253
254 final byte [] b = Bytes.toBytes("b");
255
256
257 KeyValue kv = new KeyValue(b, b, b, 2000, b);
258 KeyValue original = kv.clone();
259 writer.write(new ImmutableBytesWritable(), kv);
260 assertEquals(original,kv);
261
262
263 kv = new KeyValue(b, b, b, 1000, b);
264 original = kv.clone();
265 writer.write(new ImmutableBytesWritable(), kv);
266 assertEquals(original, kv);
267
268
269 writer.close(context);
270
271
272
273
274 FileSystem fs = FileSystem.get(conf);
275 Path attemptDirectory = hof.getDefaultWorkFile(context, "").getParent();
276 FileStatus[] sub1 = fs.listStatus(attemptDirectory);
277 FileStatus[] file = fs.listStatus(sub1[0].getPath());
278
279
280 HFile.Reader rd = HFile.createReader(fs, file[0].getPath(),
281 new CacheConfig(conf), conf);
282 Map<byte[],byte[]> finfo = rd.loadFileInfo();
283 byte[] range = finfo.get("TIMERANGE".getBytes());
284 assertNotNull(range);
285
286
287 TimeRangeTracker timeRangeTracker = new TimeRangeTracker();
288 Writables.copyWritable(range, timeRangeTracker);
289 LOG.info(timeRangeTracker.getMin() + "...." + timeRangeTracker.getMax());
290 assertEquals(1000, timeRangeTracker.getMin());
291 assertEquals(2000, timeRangeTracker.getMax());
292 rd.close();
293 } finally {
294 if (writer != null && context != null) writer.close(context);
295 dir.getFileSystem(conf).delete(dir, true);
296 }
297 }
298
299
300
301
302 @Test
303 public void testWritingPEData() throws Exception {
304 Configuration conf = util.getConfiguration();
305 Path testDir = util.getDataTestDirOnTestFS("testWritingPEData");
306 FileSystem fs = testDir.getFileSystem(conf);
307
308
309 conf.setInt("mapreduce.task.io.sort.mb", 20);
310
311 conf.setLong(HConstants.HREGION_MAX_FILESIZE, 64 * 1024);
312
313 Job job = new Job(conf, "testWritingPEData");
314 setupRandomGeneratorMapper(job);
315
316
317 byte[] startKey = new byte[RandomKVGeneratingMapper.KEYLEN_DEFAULT];
318 byte[] endKey = new byte[RandomKVGeneratingMapper.KEYLEN_DEFAULT];
319
320 Arrays.fill(startKey, (byte)0);
321 Arrays.fill(endKey, (byte)0xff);
322
323 job.setPartitionerClass(SimpleTotalOrderPartitioner.class);
324
325 SimpleTotalOrderPartitioner.setStartKey(job.getConfiguration(), startKey);
326 SimpleTotalOrderPartitioner.setEndKey(job.getConfiguration(), endKey);
327 job.setReducerClass(KeyValueSortReducer.class);
328 job.setOutputFormatClass(HFileOutputFormat.class);
329 job.setNumReduceTasks(4);
330 job.getConfiguration().setStrings("io.serializations", conf.get("io.serializations"),
331 MutationSerialization.class.getName(), ResultSerialization.class.getName(),
332 KeyValueSerialization.class.getName());
333
334 FileOutputFormat.setOutputPath(job, testDir);
335 assertTrue(job.waitForCompletion(false));
336 FileStatus [] files = fs.listStatus(testDir);
337 assertTrue(files.length > 0);
338 }
339
340 @Test
341 public void testJobConfiguration() throws Exception {
342 Configuration conf = new Configuration(this.util.getConfiguration());
343 conf.set("hbase.fs.tmp.dir", util.getDataTestDir("testJobConfiguration").toString());
344 Job job = new Job(conf);
345 job.setWorkingDirectory(util.getDataTestDir("testJobConfiguration"));
346 HTableDescriptor tableDescriptor = Mockito.mock(HTableDescriptor.class);
347 RegionLocator regionLocator = Mockito.mock(RegionLocator.class);
348 setupMockStartKeys(regionLocator);
349 setupMockTableName(regionLocator);
350 HFileOutputFormat2.configureIncrementalLoad(job, tableDescriptor, regionLocator);
351 assertEquals(job.getNumReduceTasks(), 4);
352 }
353
354 private byte [][] generateRandomStartKeys(int numKeys) {
355 Random random = new Random();
356 byte[][] ret = new byte[numKeys][];
357
358 ret[0] = HConstants.EMPTY_BYTE_ARRAY;
359 for (int i = 1; i < numKeys; i++) {
360 ret[i] =
361 PerformanceEvaluation.generateData(random, PerformanceEvaluation.DEFAULT_VALUE_LENGTH);
362 }
363 return ret;
364 }
365
366 private byte[][] generateRandomSplitKeys(int numKeys) {
367 Random random = new Random();
368 byte[][] ret = new byte[numKeys][];
369 for (int i = 0; i < numKeys; i++) {
370 ret[i] =
371 PerformanceEvaluation.generateData(random, PerformanceEvaluation.DEFAULT_VALUE_LENGTH);
372 }
373 return ret;
374 }
375
376 @Test
377 public void testMRIncrementalLoad() throws Exception {
378 LOG.info("\nStarting test testMRIncrementalLoad\n");
379 doIncrementalLoadTest(false);
380 }
381
382 @Test
383 public void testMRIncrementalLoadWithSplit() throws Exception {
384 LOG.info("\nStarting test testMRIncrementalLoadWithSplit\n");
385 doIncrementalLoadTest(true);
386 }
387
388 private void doIncrementalLoadTest(
389 boolean shouldChangeRegions) throws Exception {
390 util = new HBaseTestingUtility();
391 Configuration conf = util.getConfiguration();
392 byte[][] splitKeys = generateRandomSplitKeys(4);
393 HBaseAdmin admin = null;
394 try {
395 util.setJobWithoutMRCluster();
396 util.startMiniCluster();
397 Path testDir = util.getDataTestDirOnTestFS("testLocalMRIncrementalLoad");
398 admin = util.getHBaseAdmin();
399 HTable table = util.createTable(TABLE_NAME, FAMILIES, splitKeys);
400 assertEquals("Should start with empty table",
401 0, util.countRows(table));
402 int numRegions = -1;
403 try(RegionLocator r = table.getRegionLocator()) {
404 numRegions = r.getStartKeys().length;
405 }
406 assertEquals("Should make 5 regions", numRegions, 5);
407
408
409 runIncrementalPELoad(conf, table, testDir);
410
411 assertEquals("HFOF should not touch actual table",
412 0, util.countRows(table));
413
414
415
416 int dir = 0;
417 for (FileStatus f : testDir.getFileSystem(conf).listStatus(testDir)) {
418 for (byte[] family : FAMILIES) {
419 if (Bytes.toString(family).equals(f.getPath().getName())) {
420 ++dir;
421 }
422 }
423 }
424 assertEquals("Column family not found in FS.", FAMILIES.length, dir);
425
426
427 if (shouldChangeRegions) {
428 LOG.info("Changing regions in table");
429 admin.disableTable(table.getTableName());
430 while(util.getMiniHBaseCluster().getMaster().getAssignmentManager().
431 getRegionStates().isRegionsInTransition()) {
432 Threads.sleep(200);
433 LOG.info("Waiting on table to finish disabling");
434 }
435 util.deleteTable(table.getName());
436 byte[][] newSplitKeys = generateRandomSplitKeys(14);
437 table = util.createTable(TABLE_NAME, FAMILIES, newSplitKeys);
438 while (table.getRegionLocations().size() != 15 ||
439 !admin.isTableAvailable(table.getTableName())) {
440 Thread.sleep(200);
441 LOG.info("Waiting for new region assignment to happen");
442 }
443 }
444
445
446 new LoadIncrementalHFiles(conf).doBulkLoad(testDir, table);
447
448
449 int expectedRows = NMapInputFormat.getNumMapTasks(conf) * ROWSPERSPLIT;
450 assertEquals("LoadIncrementalHFiles should put expected data in table",
451 expectedRows, util.countRows(table));
452 Scan scan = new Scan();
453 ResultScanner results = table.getScanner(scan);
454 for (Result res : results) {
455 assertEquals(FAMILIES.length, res.rawCells().length);
456 Cell first = res.rawCells()[0];
457 for (Cell kv : res.rawCells()) {
458 assertTrue(CellUtil.matchingRow(first, kv));
459 assertTrue(Bytes.equals(CellUtil.cloneValue(first), CellUtil.cloneValue(kv)));
460 }
461 }
462 results.close();
463 String tableDigestBefore = util.checksumRows(table);
464
465
466 admin.disableTable(TABLE_NAME);
467 while (!admin.isTableDisabled(TABLE_NAME)) {
468 Thread.sleep(200);
469 LOG.info("Waiting for table to disable");
470 }
471 admin.enableTable(TABLE_NAME);
472 util.waitTableAvailable(TABLE_NAME);
473 assertEquals("Data should remain after reopening of regions",
474 tableDigestBefore, util.checksumRows(table));
475 } finally {
476 if (admin != null) admin.close();
477 util.shutdownMiniCluster();
478 }
479 }
480
481 private void runIncrementalPELoad(
482 Configuration conf, HTable table, Path outDir)
483 throws Exception {
484 Job job = new Job(conf, "testLocalMRIncrementalLoad");
485 job.setWorkingDirectory(util.getDataTestDirOnTestFS("runIncrementalPELoad"));
486 job.getConfiguration().setStrings("io.serializations", conf.get("io.serializations"),
487 MutationSerialization.class.getName(), ResultSerialization.class.getName(),
488 KeyValueSerialization.class.getName());
489 setupRandomGeneratorMapper(job);
490 HFileOutputFormat2.configureIncrementalLoad(job, table.getTableDescriptor(),
491 table.getRegionLocator());
492 FileOutputFormat.setOutputPath(job, outDir);
493
494 Assert.assertFalse( util.getTestFileSystem().exists(outDir)) ;
495
496 assertEquals(table.getRegionLocator().getAllRegionLocations().size(), job.getNumReduceTasks());
497
498 assertTrue(job.waitForCompletion(true));
499 }
500
501
502
503
504
505
506
507
508
509
510 @Test
511 public void testSerializeDeserializeFamilyCompressionMap() throws IOException {
512 for (int numCfs = 0; numCfs <= 3; numCfs++) {
513 Configuration conf = new Configuration(this.util.getConfiguration());
514 Map<String, Compression.Algorithm> familyToCompression =
515 getMockColumnFamiliesForCompression(numCfs);
516 Table table = Mockito.mock(HTable.class);
517 setupMockColumnFamiliesForCompression(table, familyToCompression);
518 HFileOutputFormat.configureCompression(table, conf);
519
520
521 Map<byte[], Algorithm> retrievedFamilyToCompressionMap = HFileOutputFormat
522 .createFamilyCompressionMap(conf);
523
524
525
526 for (Entry<String, Algorithm> entry : familyToCompression.entrySet()) {
527 assertEquals("Compression configuration incorrect for column family:"
528 + entry.getKey(), entry.getValue(),
529 retrievedFamilyToCompressionMap.get(entry.getKey().getBytes()));
530 }
531 }
532 }
533
534 private void setupMockColumnFamiliesForCompression(Table table,
535 Map<String, Compression.Algorithm> familyToCompression) throws IOException {
536 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
537 for (Entry<String, Compression.Algorithm> entry : familyToCompression.entrySet()) {
538 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
539 .setMaxVersions(1)
540 .setCompressionType(entry.getValue())
541 .setBlockCacheEnabled(false)
542 .setTimeToLive(0));
543 }
544 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
545 }
546
547
548
549
550
551 private Map<String, Compression.Algorithm>
552 getMockColumnFamiliesForCompression (int numCfs) {
553 Map<String, Compression.Algorithm> familyToCompression = new HashMap<String, Compression.Algorithm>();
554
555 if (numCfs-- > 0) {
556 familyToCompression.put("Family1!@#!@#&", Compression.Algorithm.LZO);
557 }
558 if (numCfs-- > 0) {
559 familyToCompression.put("Family2=asdads&!AASD", Compression.Algorithm.SNAPPY);
560 }
561 if (numCfs-- > 0) {
562 familyToCompression.put("Family2=asdads&!AASD", Compression.Algorithm.GZ);
563 }
564 if (numCfs-- > 0) {
565 familyToCompression.put("Family3", Compression.Algorithm.NONE);
566 }
567 return familyToCompression;
568 }
569
570
571
572
573
574
575
576
577
578
579
580 @Test
581 public void testSerializeDeserializeFamilyBloomTypeMap() throws IOException {
582 for (int numCfs = 0; numCfs <= 2; numCfs++) {
583 Configuration conf = new Configuration(this.util.getConfiguration());
584 Map<String, BloomType> familyToBloomType =
585 getMockColumnFamiliesForBloomType(numCfs);
586 Table table = Mockito.mock(HTable.class);
587 setupMockColumnFamiliesForBloomType(table,
588 familyToBloomType);
589 HFileOutputFormat.configureBloomType(table, conf);
590
591
592
593 Map<byte[], BloomType> retrievedFamilyToBloomTypeMap =
594 HFileOutputFormat
595 .createFamilyBloomTypeMap(conf);
596
597
598
599 for (Entry<String, BloomType> entry : familyToBloomType.entrySet()) {
600 assertEquals("BloomType configuration incorrect for column family:"
601 + entry.getKey(), entry.getValue(),
602 retrievedFamilyToBloomTypeMap.get(entry.getKey().getBytes()));
603 }
604 }
605 }
606
607 private void setupMockColumnFamiliesForBloomType(Table table,
608 Map<String, BloomType> familyToDataBlockEncoding) throws IOException {
609 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
610 for (Entry<String, BloomType> entry : familyToDataBlockEncoding.entrySet()) {
611 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
612 .setMaxVersions(1)
613 .setBloomFilterType(entry.getValue())
614 .setBlockCacheEnabled(false)
615 .setTimeToLive(0));
616 }
617 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
618 }
619
620
621
622
623
624 private Map<String, BloomType>
625 getMockColumnFamiliesForBloomType (int numCfs) {
626 Map<String, BloomType> familyToBloomType =
627 new HashMap<String, BloomType>();
628
629 if (numCfs-- > 0) {
630 familyToBloomType.put("Family1!@#!@#&", BloomType.ROW);
631 }
632 if (numCfs-- > 0) {
633 familyToBloomType.put("Family2=asdads&!AASD",
634 BloomType.ROWCOL);
635 }
636 if (numCfs-- > 0) {
637 familyToBloomType.put("Family3", BloomType.NONE);
638 }
639 return familyToBloomType;
640 }
641
642
643
644
645
646
647
648
649
650
651 @Test
652 public void testSerializeDeserializeFamilyBlockSizeMap() throws IOException {
653 for (int numCfs = 0; numCfs <= 3; numCfs++) {
654 Configuration conf = new Configuration(this.util.getConfiguration());
655 Map<String, Integer> familyToBlockSize =
656 getMockColumnFamiliesForBlockSize(numCfs);
657 Table table = Mockito.mock(HTable.class);
658 setupMockColumnFamiliesForBlockSize(table,
659 familyToBlockSize);
660 HFileOutputFormat.configureBlockSize(table, conf);
661
662
663
664 Map<byte[], Integer> retrievedFamilyToBlockSizeMap =
665 HFileOutputFormat
666 .createFamilyBlockSizeMap(conf);
667
668
669
670 for (Entry<String, Integer> entry : familyToBlockSize.entrySet()
671 ) {
672 assertEquals("BlockSize configuration incorrect for column family:"
673 + entry.getKey(), entry.getValue(),
674 retrievedFamilyToBlockSizeMap.get(entry.getKey().getBytes()));
675 }
676 }
677 }
678
679 private void setupMockColumnFamiliesForBlockSize(Table table,
680 Map<String, Integer> familyToDataBlockEncoding) throws IOException {
681 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
682 for (Entry<String, Integer> entry : familyToDataBlockEncoding.entrySet()) {
683 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
684 .setMaxVersions(1)
685 .setBlocksize(entry.getValue())
686 .setBlockCacheEnabled(false)
687 .setTimeToLive(0));
688 }
689 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
690 }
691
692
693
694
695
696 private Map<String, Integer>
697 getMockColumnFamiliesForBlockSize (int numCfs) {
698 Map<String, Integer> familyToBlockSize =
699 new HashMap<String, Integer>();
700
701 if (numCfs-- > 0) {
702 familyToBlockSize.put("Family1!@#!@#&", 1234);
703 }
704 if (numCfs-- > 0) {
705 familyToBlockSize.put("Family2=asdads&!AASD",
706 Integer.MAX_VALUE);
707 }
708 if (numCfs-- > 0) {
709 familyToBlockSize.put("Family2=asdads&!AASD",
710 Integer.MAX_VALUE);
711 }
712 if (numCfs-- > 0) {
713 familyToBlockSize.put("Family3", 0);
714 }
715 return familyToBlockSize;
716 }
717
718
719
720
721
722
723
724
725
726
727 @Test
728 public void testSerializeDeserializeFamilyDataBlockEncodingMap() throws IOException {
729 for (int numCfs = 0; numCfs <= 3; numCfs++) {
730 Configuration conf = new Configuration(this.util.getConfiguration());
731 Map<String, DataBlockEncoding> familyToDataBlockEncoding =
732 getMockColumnFamiliesForDataBlockEncoding(numCfs);
733 Table table = Mockito.mock(HTable.class);
734 setupMockColumnFamiliesForDataBlockEncoding(table,
735 familyToDataBlockEncoding);
736 HFileOutputFormat.configureDataBlockEncoding(table, conf);
737
738
739
740 Map<byte[], DataBlockEncoding> retrievedFamilyToDataBlockEncodingMap =
741 HFileOutputFormat
742 .createFamilyDataBlockEncodingMap(conf);
743
744
745
746 for (Entry<String, DataBlockEncoding> entry : familyToDataBlockEncoding.entrySet()) {
747 assertEquals("DataBlockEncoding configuration incorrect for column family:"
748 + entry.getKey(), entry.getValue(),
749 retrievedFamilyToDataBlockEncodingMap.get(entry.getKey().getBytes()));
750 }
751 }
752 }
753
754 private void setupMockColumnFamiliesForDataBlockEncoding(Table table,
755 Map<String, DataBlockEncoding> familyToDataBlockEncoding) throws IOException {
756 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
757 for (Entry<String, DataBlockEncoding> entry : familyToDataBlockEncoding.entrySet()) {
758 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
759 .setMaxVersions(1)
760 .setDataBlockEncoding(entry.getValue())
761 .setBlockCacheEnabled(false)
762 .setTimeToLive(0));
763 }
764 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
765 }
766
767
768
769
770
771 private Map<String, DataBlockEncoding>
772 getMockColumnFamiliesForDataBlockEncoding (int numCfs) {
773 Map<String, DataBlockEncoding> familyToDataBlockEncoding =
774 new HashMap<String, DataBlockEncoding>();
775
776 if (numCfs-- > 0) {
777 familyToDataBlockEncoding.put("Family1!@#!@#&", DataBlockEncoding.DIFF);
778 }
779 if (numCfs-- > 0) {
780 familyToDataBlockEncoding.put("Family2=asdads&!AASD",
781 DataBlockEncoding.FAST_DIFF);
782 }
783 if (numCfs-- > 0) {
784 familyToDataBlockEncoding.put("Family2=asdads&!AASD",
785 DataBlockEncoding.PREFIX);
786 }
787 if (numCfs-- > 0) {
788 familyToDataBlockEncoding.put("Family3", DataBlockEncoding.NONE);
789 }
790 return familyToDataBlockEncoding;
791 }
792
793 private void setupMockStartKeys(RegionLocator regionLocator) throws IOException {
794 byte[][] mockKeys = new byte[][] {
795 HConstants.EMPTY_BYTE_ARRAY,
796 Bytes.toBytes("aaa"),
797 Bytes.toBytes("ggg"),
798 Bytes.toBytes("zzz")
799 };
800 Mockito.doReturn(mockKeys).when(regionLocator).getStartKeys();
801 }
802
803 private void setupMockTableName(RegionLocator table) throws IOException {
804 TableName mockTableName = TableName.valueOf("mock_table");
805 Mockito.doReturn(mockTableName).when(table).getName();
806 }
807
808
809
810
811
812 @Test
813 public void testColumnFamilySettings() throws Exception {
814 Configuration conf = new Configuration(this.util.getConfiguration());
815 RecordWriter<ImmutableBytesWritable, KeyValue> writer = null;
816 TaskAttemptContext context = null;
817 Path dir = util.getDataTestDir("testColumnFamilySettings");
818
819
820 HTable table = Mockito.mock(HTable.class);
821 RegionLocator regionLocator = Mockito.mock(RegionLocator.class);
822 HTableDescriptor htd = new HTableDescriptor(TABLE_NAME);
823 Mockito.doReturn(htd).when(table).getTableDescriptor();
824 for (HColumnDescriptor hcd: this.util.generateColumnDescriptors()) {
825 htd.addFamily(hcd);
826 }
827
828
829 setupMockStartKeys(regionLocator);
830
831 try {
832
833
834
835 conf.set("io.seqfile.compression.type", "NONE");
836 conf.set("hbase.fs.tmp.dir", dir.toString());
837
838 conf.setBoolean(HFileOutputFormat2.LOCALITY_SENSITIVE_CONF_KEY, false);
839 Job job = new Job(conf, "testLocalMRIncrementalLoad");
840 job.setWorkingDirectory(util.getDataTestDirOnTestFS("testColumnFamilySettings"));
841 setupRandomGeneratorMapper(job);
842 HFileOutputFormat2.configureIncrementalLoad(job, table.getTableDescriptor(), regionLocator);
843 FileOutputFormat.setOutputPath(job, dir);
844 context = createTestTaskAttemptContext(job);
845 HFileOutputFormat hof = new HFileOutputFormat();
846 writer = hof.getRecordWriter(context);
847
848
849 writeRandomKeyValues(writer, context, htd.getFamiliesKeys(), ROWSPERSPLIT);
850 writer.close(context);
851
852
853 FileSystem fs = dir.getFileSystem(conf);
854
855
856 hof.getOutputCommitter(context).commitTask(context);
857 hof.getOutputCommitter(context).commitJob(context);
858 FileStatus[] families = FSUtils.listStatus(fs, dir, new FSUtils.FamilyDirFilter(fs));
859 assertEquals(htd.getFamilies().size(), families.length);
860 for (FileStatus f : families) {
861 String familyStr = f.getPath().getName();
862 HColumnDescriptor hcd = htd.getFamily(Bytes.toBytes(familyStr));
863
864
865 Path dataFilePath = fs.listStatus(f.getPath())[0].getPath();
866 Reader reader = HFile.createReader(fs, dataFilePath, new CacheConfig(conf), conf);
867 Map<byte[], byte[]> fileInfo = reader.loadFileInfo();
868
869 byte[] bloomFilter = fileInfo.get(StoreFile.BLOOM_FILTER_TYPE_KEY);
870 if (bloomFilter == null) bloomFilter = Bytes.toBytes("NONE");
871 assertEquals("Incorrect bloom filter used for column family " + familyStr +
872 "(reader: " + reader + ")",
873 hcd.getBloomFilterType(), BloomType.valueOf(Bytes.toString(bloomFilter)));
874 assertEquals("Incorrect compression used for column family " + familyStr +
875 "(reader: " + reader + ")", hcd.getCompression(), reader.getFileContext().getCompression());
876 }
877 } finally {
878 dir.getFileSystem(conf).delete(dir, true);
879 }
880 }
881
882
883
884
885
886 private void writeRandomKeyValues(RecordWriter<ImmutableBytesWritable, KeyValue> writer,
887 TaskAttemptContext context, Set<byte[]> families, int numRows)
888 throws IOException, InterruptedException {
889 byte keyBytes[] = new byte[Bytes.SIZEOF_INT];
890 int valLength = 10;
891 byte valBytes[] = new byte[valLength];
892
893 int taskId = context.getTaskAttemptID().getTaskID().getId();
894 assert taskId < Byte.MAX_VALUE : "Unit tests dont support > 127 tasks!";
895 final byte [] qualifier = Bytes.toBytes("data");
896 Random random = new Random();
897 for (int i = 0; i < numRows; i++) {
898
899 Bytes.putInt(keyBytes, 0, i);
900 random.nextBytes(valBytes);
901 ImmutableBytesWritable key = new ImmutableBytesWritable(keyBytes);
902
903 for (byte[] family : families) {
904 KeyValue kv = new KeyValue(keyBytes, family, qualifier, valBytes);
905 writer.write(key, kv);
906 }
907 }
908 }
909
910
911
912
913
914
915
916 @Ignore ("Flakey: See HBASE-9051") @Test
917 public void testExcludeAllFromMinorCompaction() throws Exception {
918 Configuration conf = util.getConfiguration();
919 conf.setInt("hbase.hstore.compaction.min", 2);
920 generateRandomStartKeys(5);
921
922 try {
923 util.setJobWithoutMRCluster();
924 util.startMiniCluster();
925 final FileSystem fs = util.getDFSCluster().getFileSystem();
926 HBaseAdmin admin = new HBaseAdmin(conf);
927 HTable table = util.createTable(TABLE_NAME, FAMILIES);
928 assertEquals("Should start with empty table", 0, util.countRows(table));
929
930
931 final Path storePath = HStore.getStoreHomedir(
932 FSUtils.getTableDir(FSUtils.getRootDir(conf), TABLE_NAME),
933 admin.getTableRegions(TABLE_NAME).get(0),
934 FAMILIES[0]);
935 assertEquals(0, fs.listStatus(storePath).length);
936
937
938 conf.setBoolean("hbase.mapreduce.hfileoutputformat.compaction.exclude",
939 true);
940
941 for (int i = 0; i < 2; i++) {
942 Path testDir = util.getDataTestDirOnTestFS("testExcludeAllFromMinorCompaction_" + i);
943 runIncrementalPELoad(conf, table, testDir);
944
945 new LoadIncrementalHFiles(conf).doBulkLoad(testDir, table);
946 }
947
948
949 int expectedRows = 2 * NMapInputFormat.getNumMapTasks(conf) * ROWSPERSPLIT;
950 assertEquals("LoadIncrementalHFiles should put expected data in table",
951 expectedRows, util.countRows(table));
952
953
954 assertEquals(2, fs.listStatus(storePath).length);
955
956
957 admin.compact(TABLE_NAME.getName());
958 try {
959 quickPoll(new Callable<Boolean>() {
960 public Boolean call() throws Exception {
961 return fs.listStatus(storePath).length == 1;
962 }
963 }, 5000);
964 throw new IOException("SF# = " + fs.listStatus(storePath).length);
965 } catch (AssertionError ae) {
966
967 }
968
969
970 admin.majorCompact(TABLE_NAME.getName());
971 quickPoll(new Callable<Boolean>() {
972 public Boolean call() throws Exception {
973 return fs.listStatus(storePath).length == 1;
974 }
975 }, 5000);
976
977 } finally {
978 util.shutdownMiniCluster();
979 }
980 }
981
982 @Test
983 public void testExcludeMinorCompaction() throws Exception {
984 Configuration conf = util.getConfiguration();
985 conf.setInt("hbase.hstore.compaction.min", 2);
986 generateRandomStartKeys(5);
987
988 try {
989 util.setJobWithoutMRCluster();
990 util.startMiniCluster();
991 Path testDir = util.getDataTestDirOnTestFS("testExcludeMinorCompaction");
992 final FileSystem fs = util.getTestFileSystem();
993 HBaseAdmin admin = new HBaseAdmin(conf);
994 HTable table = util.createTable(TABLE_NAME, FAMILIES);
995 assertEquals("Should start with empty table", 0, util.countRows(table));
996
997
998 final Path storePath = HStore.getStoreHomedir(
999 FSUtils.getTableDir(FSUtils.getRootDir(conf), TABLE_NAME),
1000 admin.getTableRegions(TABLE_NAME).get(0),
1001 FAMILIES[0]);
1002 assertEquals(0, fs.listStatus(storePath).length);
1003
1004
1005 Put p = new Put(Bytes.toBytes("test"));
1006 p.add(FAMILIES[0], Bytes.toBytes("1"), Bytes.toBytes("1"));
1007 table.put(p);
1008 admin.flush(TABLE_NAME.getName());
1009 assertEquals(1, util.countRows(table));
1010 quickPoll(new Callable<Boolean>() {
1011 public Boolean call() throws Exception {
1012 return fs.listStatus(storePath).length == 1;
1013 }
1014 }, 5000);
1015
1016
1017 conf.setBoolean("hbase.mapreduce.hfileoutputformat.compaction.exclude",
1018 true);
1019 runIncrementalPELoad(conf, table, testDir);
1020
1021
1022 new LoadIncrementalHFiles(conf).doBulkLoad(testDir, table);
1023
1024
1025 int expectedRows = NMapInputFormat.getNumMapTasks(conf) * ROWSPERSPLIT;
1026 assertEquals("LoadIncrementalHFiles should put expected data in table",
1027 expectedRows + 1, util.countRows(table));
1028
1029
1030 assertEquals(2, fs.listStatus(storePath).length);
1031
1032
1033 admin.compact(TABLE_NAME.getName());
1034 try {
1035 quickPoll(new Callable<Boolean>() {
1036 public Boolean call() throws Exception {
1037 List<HRegion> regions = util.getMiniHBaseCluster().getRegions(TABLE_NAME);
1038 for (HRegion region : regions) {
1039 for (Store store : region.getStores()) {
1040 store.closeAndArchiveCompactedFiles();
1041 }
1042 }
1043 return fs.listStatus(storePath).length == 1;
1044 }
1045 }, 5000);
1046 throw new IOException("SF# = " + fs.listStatus(storePath).length);
1047 } catch (AssertionError ae) {
1048
1049 }
1050
1051
1052 admin.majorCompact(TABLE_NAME.getName());
1053 quickPoll(new Callable<Boolean>() {
1054 public Boolean call() throws Exception {
1055 List<HRegion> regions = util.getMiniHBaseCluster().getRegions(TABLE_NAME);
1056 for (HRegion region : regions) {
1057 for (Store store : region.getStores()) {
1058 store.closeAndArchiveCompactedFiles();
1059 }
1060 }
1061 return fs.listStatus(storePath).length == 1;
1062 }
1063 }, 5000);
1064
1065 } finally {
1066 util.shutdownMiniCluster();
1067 }
1068 }
1069
1070 private void quickPoll(Callable<Boolean> c, int waitMs) throws Exception {
1071 int sleepMs = 10;
1072 int retries = (int) Math.ceil(((double) waitMs) / sleepMs);
1073 while (retries-- > 0) {
1074 if (c.call().booleanValue()) {
1075 return;
1076 }
1077 Thread.sleep(sleepMs);
1078 }
1079 fail();
1080 }
1081
1082 public static void main(String args[]) throws Exception {
1083 new TestHFileOutputFormat().manualTest(args);
1084 }
1085
1086 public void manualTest(String args[]) throws Exception {
1087 Configuration conf = HBaseConfiguration.create();
1088 util = new HBaseTestingUtility(conf);
1089 if ("newtable".equals(args[0])) {
1090 TableName tname = TableName.valueOf(args[1]);
1091 byte[][] splitKeys = generateRandomSplitKeys(4);
1092 HTable table = util.createTable(tname, FAMILIES, splitKeys);
1093 } else if ("incremental".equals(args[0])) {
1094 TableName tname = TableName.valueOf(args[1]);
1095 HTable table = new HTable(conf, tname);
1096 Path outDir = new Path("incremental-out");
1097 runIncrementalPELoad(conf, table, outDir);
1098 } else {
1099 throw new RuntimeException(
1100 "usage: TestHFileOutputFormat newtable | incremental");
1101 }
1102 }
1103
1104 }
1105