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