View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.mob.filecompactions;
20  
21  import static org.junit.Assert.assertEquals;
22  
23  import java.io.IOException;
24  import java.util.ArrayList;
25  import java.util.Collections;
26  import java.util.List;
27  import java.util.Random;
28  import java.util.concurrent.ExecutorService;
29  import java.util.concurrent.RejectedExecutionException;
30  import java.util.concurrent.RejectedExecutionHandler;
31  import java.util.concurrent.SynchronousQueue;
32  import java.util.concurrent.ThreadPoolExecutor;
33  import java.util.concurrent.TimeUnit;
34  
35  import org.apache.hadoop.conf.Configuration;
36  import org.apache.hadoop.fs.FileStatus;
37  import org.apache.hadoop.fs.FileSystem;
38  import org.apache.hadoop.fs.Path;
39  import org.apache.hadoop.hbase.Cell;
40  import org.apache.hadoop.hbase.CellUtil;
41  import org.apache.hadoop.hbase.HBaseTestingUtility;
42  import org.apache.hadoop.hbase.HColumnDescriptor;
43  import org.apache.hadoop.hbase.HTableDescriptor;
44  import org.apache.hadoop.hbase.NamespaceDescriptor;
45  import org.apache.hadoop.hbase.testclassification.LargeTests;
46  import org.apache.hadoop.hbase.TableName;
47  import org.apache.hadoop.hbase.client.Admin;
48  import org.apache.hadoop.hbase.client.Delete;
49  import org.apache.hadoop.hbase.client.Durability;
50  import org.apache.hadoop.hbase.client.HTable;
51  import org.apache.hadoop.hbase.client.Put;
52  import org.apache.hadoop.hbase.client.Result;
53  import org.apache.hadoop.hbase.client.ResultScanner;
54  import org.apache.hadoop.hbase.client.Scan;
55  import org.apache.hadoop.hbase.io.HFileLink;
56  import org.apache.hadoop.hbase.mob.MobConstants;
57  import org.apache.hadoop.hbase.mob.MobUtils;
58  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState;
59  import org.apache.hadoop.hbase.regionserver.HRegion;
60  import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
61  import org.apache.hadoop.hbase.util.Bytes;
62  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
63  import org.apache.hadoop.hbase.util.Threads;
64  import org.junit.After;
65  import org.junit.AfterClass;
66  import org.junit.Before;
67  import org.junit.BeforeClass;
68  import org.junit.Test;
69  import org.junit.experimental.categories.Category;
70  
71  @Category(LargeTests.class)
72  public class TestMobFileCompactor {
73    private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
74    private Configuration conf = null;
75    private String tableNameAsString;
76    private TableName tableName;
77    private HTable hTable;
78    private Admin admin;
79    private HTableDescriptor desc;
80    private HColumnDescriptor hcd1;
81    private HColumnDescriptor hcd2;
82    private FileSystem fs;
83    private final String family1 = "family1";
84    private final String family2 = "family2";
85    private final String qf1 = "qualifier1";
86    private final String qf2 = "qualifier2";
87    private byte[] KEYS = Bytes.toBytes("012");
88    private int regionNum = KEYS.length;
89    private int delRowNum = 1;
90    private int delCellNum = 6;
91    private int cellNumPerRow = 3;
92    private int rowNumPerFile = 2;
93    private static ExecutorService pool;
94  
95    @BeforeClass
96    public static void setUpBeforeClass() throws Exception {
97      TEST_UTIL.getConfiguration().setInt("hbase.master.info.port", 0);
98      TEST_UTIL.getConfiguration().setBoolean("hbase.regionserver.info.port.auto", true);
99      TEST_UTIL.getConfiguration().setLong(MobConstants.MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD, 5000);
100     TEST_UTIL.getConfiguration().setInt("hfile.format.version", 3);
101     TEST_UTIL.startMiniCluster(1);
102     pool = createThreadPool(TEST_UTIL.getConfiguration());
103   }
104 
105   @AfterClass
106   public static void tearDownAfterClass() throws Exception {
107     pool.shutdown();
108     TEST_UTIL.shutdownMiniCluster();
109   }
110 
111   @Before
112   public void setUp() throws Exception {
113     fs = TEST_UTIL.getTestFileSystem();
114     conf = TEST_UTIL.getConfiguration();
115     long tid = System.currentTimeMillis();
116     tableNameAsString = "testMob" + tid;
117     tableName = TableName.valueOf(tableNameAsString);
118     hcd1 = new HColumnDescriptor(family1);
119     hcd1.setMobEnabled(true);
120     hcd1.setMobThreshold(0L);
121     hcd1.setMaxVersions(4);
122     hcd2 = new HColumnDescriptor(family2);
123     hcd2.setMobEnabled(true);
124     hcd2.setMobThreshold(0L);
125     hcd2.setMaxVersions(4);
126     desc = new HTableDescriptor(tableName);
127     desc.addFamily(hcd1);
128     desc.addFamily(hcd2);
129     admin = TEST_UTIL.getHBaseAdmin();
130     admin.createTable(desc, getSplitKeys());
131     hTable = new HTable(conf, tableNameAsString);
132     hTable.setAutoFlush(false, false);
133   }
134 
135   @After
136   public void tearDown() throws Exception {
137     admin.disableTable(tableName);
138     admin.deleteTable(tableName);
139     admin.close();
140     hTable.close();
141     fs.delete(TEST_UTIL.getDataTestDir(), true);
142   }
143 
144   @Test
145   public void testCompactionWithoutDelFilesWithNamespace() throws Exception {
146     resetConf();
147     // create a table with namespace
148     NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create("ns").build();
149     String tableNameAsString = "ns:testCompactionWithoutDelFilesWithNamespace";
150     admin.createNamespace(namespaceDescriptor);
151     TableName tableName = TableName.valueOf(tableNameAsString);
152     HColumnDescriptor hcd1 = new HColumnDescriptor(family1);
153     hcd1.setMobEnabled(true);
154     hcd1.setMobThreshold(0L);
155     hcd1.setMaxVersions(4);
156     HColumnDescriptor hcd2 = new HColumnDescriptor(family2);
157     hcd2.setMobEnabled(true);
158     hcd2.setMobThreshold(0L);
159     hcd2.setMaxVersions(4);
160     HTableDescriptor desc = new HTableDescriptor(tableName);
161     desc.addFamily(hcd1);
162     desc.addFamily(hcd2);
163     admin.createTable(desc, getSplitKeys());
164     HTable table = new HTable(conf, tableName);
165     table.setAutoFlush(false, false);
166 
167     int count = 4;
168     // generate mob files
169     loadData(admin, table, tableName, count, rowNumPerFile);
170     int rowNumPerRegion = count * rowNumPerFile;
171 
172     assertEquals("Before compaction: mob rows count", regionNum * rowNumPerRegion,
173       countMobRows(table));
174     assertEquals("Before compaction: mob file count", regionNum * count,
175       countFiles(tableName, true, family1));
176     assertEquals("Before compaction: del file count", 0, countFiles(tableName, false, family1));
177 
178     MobFileCompactor compactor = new PartitionedMobFileCompactor(conf, fs, tableName, hcd1, pool);
179     compactor.compact();
180 
181     assertEquals("After compaction: mob rows count", regionNum * rowNumPerRegion,
182       countMobRows(table));
183     assertEquals("After compaction: mob file count", regionNum,
184       countFiles(tableName, true, family1));
185     assertEquals("After compaction: del file count", 0, countFiles(tableName, false, family1));
186 
187     table.close();
188     admin.disableTable(tableName);
189     admin.deleteTable(tableName);
190     admin.deleteNamespace("ns");
191   }
192 
193   @Test
194   public void testCompactionWithoutDelFiles() throws Exception {
195     resetConf();
196     int count = 4;
197     // generate mob files
198     loadData(admin, hTable, tableName, count, rowNumPerFile);
199     int rowNumPerRegion = count*rowNumPerFile;
200 
201     assertEquals("Before compaction: mob rows count", regionNum*rowNumPerRegion,
202         countMobRows(hTable));
203     assertEquals("Before compaction: mob file count", regionNum * count,
204       countFiles(tableName, true, family1));
205     assertEquals("Before compaction: del file count", 0, countFiles(tableName, false, family1));
206 
207     MobFileCompactor compactor = new PartitionedMobFileCompactor(conf, fs, tableName, hcd1, pool);
208     compactor.compact();
209 
210     assertEquals("After compaction: mob rows count", regionNum*rowNumPerRegion,
211         countMobRows(hTable));
212     assertEquals("After compaction: mob file count", regionNum,
213       countFiles(tableName, true, family1));
214     assertEquals("After compaction: del file count", 0, countFiles(tableName, false, family1));
215   }
216 
217   @Test
218   public void testCompactionWithDelFiles() throws Exception {
219     resetConf();
220     int count = 4;
221     // generate mob files
222     loadData(admin, hTable, tableName, count, rowNumPerFile);
223     int rowNumPerRegion = count*rowNumPerFile;
224 
225     assertEquals("Before deleting: mob rows count", regionNum*rowNumPerRegion,
226         countMobRows(hTable));
227     assertEquals("Before deleting: mob cells count", regionNum*cellNumPerRow*rowNumPerRegion,
228         countMobCells(hTable));
229     assertEquals("Before deleting: family1 mob file count", regionNum*count,
230         countFiles(tableName, true, family1));
231     assertEquals("Before deleting: family2 mob file count", regionNum*count,
232         countFiles(tableName, true, family2));
233 
234     createDelFile();
235 
236     assertEquals("Before compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
237         countMobRows(hTable));
238     assertEquals("Before compaction: mob cells count",
239         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
240     assertEquals("Before compaction: family1 mob file count", regionNum*count,
241         countFiles(tableName, true, family1));
242     assertEquals("Before compaction: family2 file count", regionNum*count,
243         countFiles(tableName, true, family2));
244     assertEquals("Before compaction: family1 del file count", regionNum,
245         countFiles(tableName, false, family1));
246     assertEquals("Before compaction: family2 del file count", regionNum,
247         countFiles(tableName, false, family2));
248 
249     // do the mob file compaction
250     MobFileCompactor compactor = new PartitionedMobFileCompactor(conf, fs, tableName, hcd1, pool);
251     compactor.compact();
252 
253     assertEquals("After compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
254         countMobRows(hTable));
255     assertEquals("After compaction: mob cells count",
256         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
257     assertEquals("After compaction: family1 mob file count", regionNum,
258         countFiles(tableName, true, family1));
259     assertEquals("After compaction: family2 mob file count", regionNum*count,
260         countFiles(tableName, true, family2));
261     assertEquals("After compaction: family1 del file count", 0,
262       countFiles(tableName, false, family1));
263     assertEquals("After compaction: family2 del file count", regionNum,
264         countFiles(tableName, false, family2));
265     assertRefFileNameEqual(family1);
266   }
267 
268   @Test
269   public void testCompactionWithDelFilesAndNotMergeAllFiles() throws Exception {
270     resetConf();
271     int mergeSize = 5000;
272     // change the mob compaction merge size
273     conf.setLong(MobConstants.MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD, mergeSize);
274 
275     int count = 4;
276     // generate mob files
277     loadData(admin, hTable, tableName, count, rowNumPerFile);
278     int rowNumPerRegion = count*rowNumPerFile;
279 
280     assertEquals("Before deleting: mob rows count", regionNum*rowNumPerRegion,
281         countMobRows(hTable));
282     assertEquals("Before deleting: mob cells count", regionNum*cellNumPerRow*rowNumPerRegion,
283         countMobCells(hTable));
284     assertEquals("Before deleting: mob file count", regionNum * count,
285       countFiles(tableName, true, family1));
286 
287     int largeFilesCount = countLargeFiles(mergeSize, family1);
288     createDelFile();
289 
290     assertEquals("Before compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
291         countMobRows(hTable));
292     assertEquals("Before compaction: mob cells count",
293         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
294     assertEquals("Before compaction: family1 mob file count", regionNum*count,
295         countFiles(tableName, true, family1));
296     assertEquals("Before compaction: family2 mob file count", regionNum*count,
297         countFiles(tableName, true, family2));
298     assertEquals("Before compaction: family1 del file count", regionNum,
299         countFiles(tableName, false, family1));
300     assertEquals("Before compaction: family2 del file count", regionNum,
301         countFiles(tableName, false, family2));
302 
303     // do the mob file compaction
304     MobFileCompactor compactor = new PartitionedMobFileCompactor(conf, fs, tableName, hcd1, pool);
305     compactor.compact();
306 
307     assertEquals("After compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
308         countMobRows(hTable));
309     assertEquals("After compaction: mob cells count",
310         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
311     // After the compaction, the files smaller than the mob compaction merge size
312     // is merge to one file
313     assertEquals("After compaction: family1 mob file count", largeFilesCount + regionNum,
314         countFiles(tableName, true, family1));
315     assertEquals("After compaction: family2 mob file count", regionNum*count,
316         countFiles(tableName, true, family2));
317     assertEquals("After compaction: family1 del file count", regionNum,
318         countFiles(tableName, false, family1));
319     assertEquals("After compaction: family2 del file count", regionNum,
320         countFiles(tableName, false, family2));
321   }
322 
323   @Test
324   public void testCompactionWithDelFilesAndWithSmallCompactionBatchSize() throws Exception {
325     resetConf();
326     int batchSize = 2;
327     conf.setInt(MobConstants.MOB_FILE_COMPACTION_BATCH_SIZE, batchSize);
328     int count = 4;
329     // generate mob files
330     loadData(admin, hTable, tableName, count, rowNumPerFile);
331     int rowNumPerRegion = count*rowNumPerFile;
332 
333     assertEquals("Before deleting: mob row count", regionNum*rowNumPerRegion,
334         countMobRows(hTable));
335     assertEquals("Before deleting: family1 mob file count", regionNum*count,
336         countFiles(tableName, true, family1));
337     assertEquals("Before deleting: family2 mob file count", regionNum*count,
338         countFiles(tableName, true, family2));
339 
340     createDelFile();
341 
342     assertEquals("Before compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
343         countMobRows(hTable));
344     assertEquals("Before compaction: mob cells count",
345         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
346     assertEquals("Before compaction: family1 mob file count", regionNum*count,
347         countFiles(tableName, true, family1));
348     assertEquals("Before compaction: family2 mob file count", regionNum*count,
349         countFiles(tableName, true, family2));
350     assertEquals("Before compaction: family1 del file count", regionNum,
351         countFiles(tableName, false, family1));
352     assertEquals("Before compaction: family2 del file count", regionNum,
353         countFiles(tableName, false, family2));
354 
355     // do the mob file compaction
356     MobFileCompactor compactor = new PartitionedMobFileCompactor(conf, fs, tableName, hcd1, pool);
357     compactor.compact();
358 
359     assertEquals("After compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
360         countMobRows(hTable));
361     assertEquals("After compaction: mob cells count",
362         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
363     assertEquals("After compaction: family1 mob file count", regionNum*(count/batchSize),
364         countFiles(tableName, true, family1));
365     assertEquals("After compaction: family2 mob file count", regionNum*count,
366         countFiles(tableName, true, family2));
367     assertEquals("After compaction: family1 del file count", 0,
368       countFiles(tableName, false, family1));
369     assertEquals("After compaction: family2 del file count", regionNum,
370         countFiles(tableName, false, family2));
371   }
372 
373   @Test
374   public void testCompactionWithHFileLink() throws IOException, InterruptedException {
375     resetConf();
376     int count = 4;
377     // generate mob files
378     loadData(admin, hTable, tableName, count, rowNumPerFile);
379     int rowNumPerRegion = count*rowNumPerFile;
380 
381     long tid = System.currentTimeMillis();
382     byte[] snapshotName1 = Bytes.toBytes("snaptb-" + tid);
383     // take a snapshot
384     admin.snapshot(snapshotName1, tableName);
385 
386     createDelFile();
387 
388     assertEquals("Before compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
389         countMobRows(hTable));
390     assertEquals("Before compaction: mob cells count",
391         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
392     assertEquals("Before compaction: family1 mob file count", regionNum*count,
393         countFiles(tableName, true, family1));
394     assertEquals("Before compaction: family2 mob file count", regionNum*count,
395         countFiles(tableName, true, family2));
396     assertEquals("Before compaction: family1 del file count", regionNum,
397         countFiles(tableName, false, family1));
398     assertEquals("Before compaction: family2 del file count", regionNum,
399         countFiles(tableName, false, family2));
400 
401     // do the mob file compaction
402     MobFileCompactor compactor = new PartitionedMobFileCompactor(conf, fs, tableName, hcd1, pool);
403     compactor.compact();
404 
405     assertEquals("After first compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
406         countMobRows(hTable));
407     assertEquals("After first compaction: mob cells count",
408         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
409     assertEquals("After first compaction: family1 mob file count", regionNum,
410         countFiles(tableName, true, family1));
411     assertEquals("After first compaction: family2 mob file count", regionNum*count,
412         countFiles(tableName, true, family2));
413     assertEquals("After first compaction: family1 del file count", 0,
414       countFiles(tableName, false, family1));
415     assertEquals("After first compaction: family2 del file count", regionNum,
416         countFiles(tableName, false, family2));
417     assertEquals("After first compaction: family1 hfilelink count", 0, countHFileLinks(family1));
418     assertEquals("After first compaction: family2 hfilelink count", 0, countHFileLinks(family2));
419 
420     admin.disableTable(tableName);
421     // Restore from snapshot, the hfilelink will exist in mob dir
422     admin.restoreSnapshot(snapshotName1);
423     admin.enableTable(tableName);
424 
425     assertEquals("After restoring snapshot: mob rows count", regionNum*rowNumPerRegion,
426         countMobRows(hTable));
427     assertEquals("After restoring snapshot: mob cells count",
428         regionNum*cellNumPerRow*rowNumPerRegion, countMobCells(hTable));
429     assertEquals("After restoring snapshot: family1 mob file count", regionNum*count,
430         countFiles(tableName, true, family1));
431     assertEquals("After restoring snapshot: family2 mob file count", regionNum*count,
432         countFiles(tableName, true, family2));
433     assertEquals("After restoring snapshot: family1 del file count", 0,
434         countFiles(tableName, false, family1));
435     assertEquals("After restoring snapshot: family2 del file count", 0,
436         countFiles(tableName, false, family2));
437     assertEquals("After restoring snapshot: family1 hfilelink count", regionNum*count,
438         countHFileLinks(family1));
439     assertEquals("After restoring snapshot: family2 hfilelink count", 0,
440         countHFileLinks(family2));
441 
442     compactor.compact();
443 
444     assertEquals("After second compaction: mob rows count", regionNum*rowNumPerRegion,
445         countMobRows(hTable));
446     assertEquals("After second compaction: mob cells count",
447         regionNum*cellNumPerRow*rowNumPerRegion, countMobCells(hTable));
448     assertEquals("After second compaction: family1 mob file count", regionNum,
449         countFiles(tableName, true, family1));
450     assertEquals("After second compaction: family2 mob file count", regionNum*count,
451         countFiles(tableName, true, family2));
452     assertEquals("After second compaction: family1 del file count", 0,
453       countFiles(tableName, false, family1));
454     assertEquals("After second compaction: family2 del file count", 0,
455       countFiles(tableName, false, family2));
456     assertEquals("After second compaction: family1 hfilelink count", 0, countHFileLinks(family1));
457     assertEquals("After second compaction: family2 hfilelink count", 0, countHFileLinks(family2));
458     assertRefFileNameEqual(family1);
459   }
460 
461   @Test
462   public void testCompactionFromAdmin() throws Exception {
463     int count = 4;
464     // generate mob files
465     loadData(admin, hTable, tableName, count, rowNumPerFile);
466     int rowNumPerRegion = count*rowNumPerFile;
467 
468     assertEquals("Before deleting: mob rows count", regionNum*rowNumPerRegion,
469         countMobRows(hTable));
470     assertEquals("Before deleting: mob cells count", regionNum*cellNumPerRow*rowNumPerRegion,
471         countMobCells(hTable));
472     assertEquals("Before deleting: family1 mob file count", regionNum*count,
473         countFiles(tableName, true, family1));
474     assertEquals("Before deleting: family2 mob file count", regionNum*count,
475         countFiles(tableName, true, family2));
476 
477     createDelFile();
478 
479     assertEquals("Before compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
480         countMobRows(hTable));
481     assertEquals("Before compaction: mob cells count",
482         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
483     assertEquals("Before compaction: family1 mob file count", regionNum*count,
484         countFiles(tableName, true, family1));
485     assertEquals("Before compaction: family2 file count", regionNum*count,
486         countFiles(tableName, true, family2));
487     assertEquals("Before compaction: family1 del file count", regionNum,
488         countFiles(tableName, false, family1));
489     assertEquals("Before compaction: family2 del file count", regionNum,
490         countFiles(tableName, false, family2));
491 
492     int largeFilesCount = countLargeFiles(5000, family1);
493     // do the mob file compaction
494     admin.compactMob(tableName, hcd1.getName());
495 
496     waitUntilCompactionFinished(tableName);
497     assertEquals("After compaction: mob rows count", regionNum * (rowNumPerRegion - delRowNum),
498       countMobRows(hTable));
499     assertEquals("After compaction: mob cells count", regionNum
500       * (cellNumPerRow * rowNumPerRegion - delCellNum), countMobCells(hTable));
501     assertEquals("After compaction: family1 mob file count", regionNum + largeFilesCount,
502       countFiles(tableName, true, family1));
503     assertEquals("After compaction: family2 mob file count", regionNum * count,
504       countFiles(tableName, true, family2));
505     assertEquals("After compaction: family1 del file count", regionNum,
506       countFiles(tableName, false, family1));
507     assertEquals("After compaction: family2 del file count", regionNum,
508       countFiles(tableName, false, family2));
509     assertRefFileNameEqual(family1);
510   }
511 
512   @Test
513   public void testMajorCompactionFromAdmin() throws Exception {
514     int count = 4;
515     // generate mob files
516     loadData(admin, hTable, tableName, count, rowNumPerFile);
517     int rowNumPerRegion = count*rowNumPerFile;
518 
519     assertEquals("Before deleting: mob rows count", regionNum*rowNumPerRegion,
520         countMobRows(hTable));
521     assertEquals("Before deleting: mob cells count", regionNum*cellNumPerRow*rowNumPerRegion,
522         countMobCells(hTable));
523     assertEquals("Before deleting: mob file count", regionNum*count,
524         countFiles(tableName, true, family1));
525 
526     createDelFile();
527 
528     assertEquals("Before compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
529         countMobRows(hTable));
530     assertEquals("Before compaction: mob cells count",
531         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
532     assertEquals("Before compaction: family1 mob file count", regionNum*count,
533         countFiles(tableName, true, family1));
534     assertEquals("Before compaction: family2 mob file count", regionNum*count,
535         countFiles(tableName, true, family2));
536     assertEquals("Before compaction: family1 del file count", regionNum,
537         countFiles(tableName, false, family1));
538     assertEquals("Before compaction: family2 del file count", regionNum,
539         countFiles(tableName, false, family2));
540 
541     // do the major mob file compaction, it will force all files to compaction
542     admin.majorCompactMob(tableName, hcd1.getName());
543 
544     waitUntilCompactionFinished(tableName);
545     assertEquals("After compaction: mob rows count", regionNum*(rowNumPerRegion-delRowNum),
546         countMobRows(hTable));
547     assertEquals("After compaction: mob cells count",
548         regionNum*(cellNumPerRow*rowNumPerRegion-delCellNum), countMobCells(hTable));
549     assertEquals("After compaction: family1 mob file count", regionNum,
550         countFiles(tableName, true, family1));
551     assertEquals("After compaction: family2 mob file count", regionNum*count,
552         countFiles(tableName, true, family2));
553     assertEquals("After compaction: family1 del file count", 0,
554         countFiles(tableName, false, family1));
555     assertEquals("After compaction: family2 del file count", regionNum,
556         countFiles(tableName, false, family2));
557   }
558 
559   private void waitUntilCompactionFinished(TableName tableName) throws IOException,
560     InterruptedException {
561     long finished = EnvironmentEdgeManager.currentTime() + 60000;
562     CompactionState state = admin.getMobCompactionState(tableName);
563     while (EnvironmentEdgeManager.currentTime() < finished) {
564       if (state == CompactionState.NONE) {
565         break;
566       }
567       state = admin.getMobCompactionState(tableName);
568       Thread.sleep(10);
569     }
570     assertEquals(CompactionState.NONE, state);
571   }
572 
573   /**
574    * Gets the number of rows in the given table.
575    * @param table to get the  scanner
576    * @return the number of rows
577    */
578   private int countMobRows(final HTable table) throws IOException {
579     Scan scan = new Scan();
580     // Do not retrieve the mob data when scanning
581     scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
582     ResultScanner results = table.getScanner(scan);
583     int count = 0;
584     for (Result res : results) {
585       count++;
586     }
587     results.close();
588     return count;
589   }
590 
591   /**
592    * Gets the number of cells in the given table.
593    * @param table to get the  scanner
594    * @return the number of cells
595    */
596   private int countMobCells(final HTable table) throws IOException {
597     Scan scan = new Scan();
598     // Do not retrieve the mob data when scanning
599     scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
600     ResultScanner results = table.getScanner(scan);
601     int count = 0;
602     for (Result res : results) {
603       for (Cell cell : res.listCells()) {
604         count++;
605       }
606     }
607     results.close();
608     return count;
609   }
610 
611   /**
612    * Gets the number of files in the mob path.
613    * @param isMobFile gets number of the mob files or del files
614    * @param familyName the family name
615    * @return the number of the files
616    */
617   private int countFiles(TableName tableName, boolean isMobFile, String familyName)
618     throws IOException {
619     Path mobDirPath = MobUtils.getMobFamilyPath(
620         MobUtils.getMobRegionPath(conf, tableName), familyName);
621     int count = 0;
622     if (fs.exists(mobDirPath)) {
623       FileStatus[] files = fs.listStatus(mobDirPath);
624       for (FileStatus file : files) {
625         if (isMobFile == true) {
626           if (!StoreFileInfo.isDelFile(file.getPath())) {
627             count++;
628           }
629         } else {
630           if (StoreFileInfo.isDelFile(file.getPath())) {
631             count++;
632           }
633         }
634       }
635     }
636     return count;
637   }
638 
639   /**
640    * Gets the number of HFileLink in the mob path.
641    * @param familyName the family name
642    * @return the number of the HFileLink
643    */
644   private int countHFileLinks(String familyName) throws IOException {
645     Path mobDirPath = MobUtils.getMobFamilyPath(
646         MobUtils.getMobRegionPath(conf, tableName), familyName);
647     int count = 0;
648     if (fs.exists(mobDirPath)) {
649       FileStatus[] files = fs.listStatus(mobDirPath);
650       for (FileStatus file : files) {
651         if (HFileLink.isHFileLink(file.getPath())) {
652           count++;
653         }
654       }
655     }
656     return count;
657   }
658 
659   /**
660    * Gets the number of files.
661    * @param size the size of the file
662    * @param familyName the family name
663    * @return the number of files large than the size
664    */
665   private int countLargeFiles(int size, String familyName) throws IOException {
666     Path mobDirPath = MobUtils.getMobFamilyPath(
667         MobUtils.getMobRegionPath(conf, tableName), familyName);
668     int count = 0;
669     if (fs.exists(mobDirPath)) {
670       FileStatus[] files = fs.listStatus(mobDirPath);
671       for (FileStatus file : files) {
672         // ignore the del files in the mob path
673         if ((!StoreFileInfo.isDelFile(file.getPath()))
674             && (file.getLen() > size)) {
675           count++;
676         }
677       }
678     }
679     return count;
680   }
681 
682   /**
683    * loads some data to the table.
684    * @param count the mob file number
685    */
686   private void loadData(Admin admin, HTable table, TableName tableName, int fileNum,
687     int rowNumPerFile) throws IOException, InterruptedException {
688     if (fileNum <= 0) {
689       throw new IllegalArgumentException();
690     }
691     for (byte k0 : KEYS) {
692       byte[] k = new byte[] { k0 };
693       for (int i = 0; i < fileNum * rowNumPerFile; i++) {
694         byte[] key = Bytes.add(k, Bytes.toBytes(i));
695         byte[] mobVal = makeDummyData(10 * (i + 1));
696         Put put = new Put(key);
697         put.setDurability(Durability.SKIP_WAL);
698         put.add(Bytes.toBytes(family1), Bytes.toBytes(qf1), mobVal);
699         put.add(Bytes.toBytes(family1), Bytes.toBytes(qf2), mobVal);
700         put.add(Bytes.toBytes(family2), Bytes.toBytes(qf1), mobVal);
701         table.put(put);
702         if ((i + 1) % rowNumPerFile == 0) {
703           table.flushCommits();
704           admin.flush(tableName);
705         }
706       }
707     }
708   }
709 
710   /**
711    * delete the row, family and cell to create the del file
712    */
713   private void createDelFile() throws IOException, InterruptedException {
714     for (byte k0 : KEYS) {
715       byte[] k = new byte[] { k0 };
716       // delete a family
717       byte[] key1 = Bytes.add(k, Bytes.toBytes(0));
718       Delete delete1 = new Delete(key1);
719       delete1.deleteFamily(Bytes.toBytes(family1));
720       hTable.delete(delete1);
721       // delete one row
722       byte[] key2 = Bytes.add(k, Bytes.toBytes(2));
723       Delete delete2 = new Delete(key2);
724       hTable.delete(delete2);
725       // delete one cell
726       byte[] key3 = Bytes.add(k, Bytes.toBytes(4));
727       Delete delete3 = new Delete(key3);
728       delete3.deleteColumn(Bytes.toBytes(family1), Bytes.toBytes(qf1));
729       hTable.delete(delete3);
730       hTable.flushCommits();
731       admin.flush(tableName);
732       List<HRegion> regions = TEST_UTIL.getHBaseCluster().getRegions(
733           Bytes.toBytes(tableNameAsString));
734       for (HRegion region : regions) {
735         region.waitForFlushesAndCompactions();
736         region.compact(true);
737       }
738     }
739   }
740   /**
741    * Creates the dummy data with a specific size.
742    * @param the size of data
743    * @return the dummy data
744    */
745   private byte[] makeDummyData(int size) {
746     byte[] dummyData = new byte[size];
747     new Random().nextBytes(dummyData);
748     return dummyData;
749   }
750 
751   /**
752    * Gets the split keys
753    */
754   private byte[][] getSplitKeys() {
755     byte[][] splitKeys = new byte[KEYS.length - 1][];
756     for (int i = 0; i < splitKeys.length; ++i) {
757       splitKeys[i] = new byte[] { KEYS[i + 1] };
758     }
759     return splitKeys;
760   }
761 
762   private static ExecutorService createThreadPool(Configuration conf) {
763     int maxThreads = 10;
764     long keepAliveTime = 60;
765     final SynchronousQueue<Runnable> queue = new SynchronousQueue<Runnable>();
766     ThreadPoolExecutor pool = new ThreadPoolExecutor(1, maxThreads,
767         keepAliveTime, TimeUnit.SECONDS, queue,
768         Threads.newDaemonThreadFactory("MobFileCompactionChore"),
769         new RejectedExecutionHandler() {
770           @Override
771           public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
772             try {
773               // waiting for a thread to pick up instead of throwing exceptions.
774               queue.put(r);
775             } catch (InterruptedException e) {
776               throw new RejectedExecutionException(e);
777             }
778           }
779         });
780     ((ThreadPoolExecutor) pool).allowCoreThreadTimeOut(true);
781     return pool;
782   }
783 
784   private void assertRefFileNameEqual(String familyName) throws IOException {
785     Scan scan = new Scan();
786     scan.addFamily(Bytes.toBytes(familyName));
787     // Do not retrieve the mob data when scanning
788     scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
789     ResultScanner results = hTable.getScanner(scan);
790     Path mobFamilyPath = new Path(MobUtils.getMobRegionPath(TEST_UTIL.getConfiguration(),
791         tableName), familyName);
792     List<Path> actualFilePaths = new ArrayList<>();
793     List<Path> expectFilePaths = new ArrayList<>();
794     for (Result res : results) {
795       for (Cell cell : res.listCells()) {
796         byte[] referenceValue = CellUtil.cloneValue(cell);
797         String fileName = Bytes.toString(referenceValue, Bytes.SIZEOF_INT,
798             referenceValue.length - Bytes.SIZEOF_INT);
799         Path targetPath = new Path(mobFamilyPath, fileName);
800         if(!actualFilePaths.contains(targetPath)) {
801           actualFilePaths.add(targetPath);
802         }
803       }
804     }
805     results.close();
806     if (fs.exists(mobFamilyPath)) {
807       FileStatus[] files = fs.listStatus(mobFamilyPath);
808       for (FileStatus file : files) {
809         if (!StoreFileInfo.isDelFile(file.getPath())) {
810           expectFilePaths.add(file.getPath());
811         }
812       }
813     }
814     Collections.sort(actualFilePaths);
815     Collections.sort(expectFilePaths);
816     assertEquals(expectFilePaths, actualFilePaths);
817   }
818 
819   /**
820    * Resets the configuration.
821    */
822   private void resetConf() {
823     conf.setLong(MobConstants.MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD,
824       MobConstants.DEFAULT_MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD);
825     conf.setInt(MobConstants.MOB_FILE_COMPACTION_BATCH_SIZE,
826       MobConstants.DEFAULT_MOB_FILE_COMPACTION_BATCH_SIZE);
827   }
828 }