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