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