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