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