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