View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.mob.mapreduce;
19  
20  import static org.junit.Assert.assertEquals;
21  
22  import java.io.IOException;
23  import java.util.Random;
24  import java.util.Set;
25  import java.util.TreeSet;
26  
27  import org.apache.hadoop.conf.Configuration;
28  import org.apache.hadoop.fs.FileStatus;
29  import org.apache.hadoop.fs.Path;
30  import org.apache.hadoop.hbase.HBaseTestingUtility;
31  import org.apache.hadoop.hbase.HColumnDescriptor;
32  import org.apache.hadoop.hbase.HTableDescriptor;
33  import org.apache.hadoop.hbase.testclassification.MediumTests;
34  import org.apache.hadoop.hbase.NamespaceDescriptor;
35  import org.apache.hadoop.hbase.TableName;
36  import org.apache.hadoop.hbase.client.Admin;
37  import org.apache.hadoop.hbase.client.HTable;
38  import org.apache.hadoop.hbase.client.Put;
39  import org.apache.hadoop.hbase.client.Result;
40  import org.apache.hadoop.hbase.client.ResultScanner;
41  import org.apache.hadoop.hbase.client.Scan;
42  import org.apache.hadoop.hbase.mob.MobConstants;
43  import org.apache.hadoop.hbase.mob.MobUtils;
44  import org.apache.hadoop.hbase.util.Bytes;
45  import org.apache.hadoop.util.ToolRunner;
46  import org.junit.After;
47  import org.junit.AfterClass;
48  import org.junit.Before;
49  import org.junit.BeforeClass;
50  import org.junit.Test;
51  import org.junit.experimental.categories.Category;
52  
53  @Category(MediumTests.class)
54  public class TestMobSweeper {
55    private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
56    private String tableName;
57    private final static String row = "row_";
58    private final static String family = "family";
59    private final static String column = "column";
60    private static HTable table;
61    private static Admin admin;
62  
63    private Random random = new Random();
64    @BeforeClass
65    public static void setUpBeforeClass() throws Exception {
66      TEST_UTIL.getConfiguration().setInt("hbase.master.info.port", 0);
67      TEST_UTIL.getConfiguration().setBoolean("hbase.regionserver.info.port.auto", true);
68      TEST_UTIL.getConfiguration().setInt("hbase.hstore.compaction.min", 15); // avoid major compactions
69      TEST_UTIL.getConfiguration().setInt("hbase.hstore.compaction.max", 30); // avoid major compactions
70      TEST_UTIL.getConfiguration().setInt("hfile.format.version", 3);
71  
72      TEST_UTIL.startMiniCluster();
73  
74      TEST_UTIL.startMiniMapReduceCluster();
75    }
76  
77    @AfterClass
78    public static void tearDownAfterClass() throws Exception {
79      TEST_UTIL.shutdownMiniCluster();
80      TEST_UTIL.shutdownMiniMapReduceCluster();
81    }
82  
83    @SuppressWarnings("deprecation")
84    @Before
85    public void setUp() throws Exception {
86      long tid = System.currentTimeMillis();
87      tableName = "testSweeper" + tid;
88      HTableDescriptor desc = new HTableDescriptor(tableName);
89      HColumnDescriptor hcd = new HColumnDescriptor(family);
90      hcd.setMobEnabled(true);
91      hcd.setMobThreshold(3L);
92      hcd.setMaxVersions(4);
93      desc.addFamily(hcd);
94  
95      admin = TEST_UTIL.getHBaseAdmin();
96      admin.createTable(desc);
97      table = new HTable(TEST_UTIL.getConfiguration(), tableName);
98      table.setAutoFlush(false, false);
99  
100   }
101 
102   @After
103   public void tearDown() throws Exception {
104     admin.disableTable(TableName.valueOf(tableName));
105     admin.deleteTable(TableName.valueOf(tableName));
106     admin.close();
107   }
108 
109   private Path getMobFamilyPath(Configuration conf, String tableNameStr,
110                                 String familyName) {
111     Path p = new Path(MobUtils.getMobRegionPath(conf, TableName.valueOf(tableNameStr)),
112             familyName);
113     return p;
114   }
115 
116   private String mergeString(Set<String> set) {
117     StringBuilder sb = new StringBuilder();
118     for (String s : set)
119       sb.append(s);
120     return sb.toString();
121   }
122 
123   private void generateMobTable(Admin admin, HTable table, String tableName, int count,
124     int flushStep) throws IOException, InterruptedException {
125     if (count <= 0 || flushStep <= 0)
126       return;
127     int index = 0;
128     for (int i = 0; i < count; i++) {
129       byte[] mobVal = new byte[101*1024];
130       random.nextBytes(mobVal);
131 
132       Put put = new Put(Bytes.toBytes(row + i));
133       put.add(Bytes.toBytes(family), Bytes.toBytes(column), mobVal);
134       table.put(put);
135       if (index++ % flushStep == 0) {
136         table.flushCommits();
137         admin.flush(TableName.valueOf(tableName));
138       }
139     }
140     table.flushCommits();
141     admin.flush(TableName.valueOf(tableName));
142   }
143 
144   @Test
145   public void testSweeper() throws Exception {
146     int count = 10;
147     //create table and generate 10 mob files
148     generateMobTable(admin, table, tableName, count, 1);
149     //get mob files
150     Path mobFamilyPath = getMobFamilyPath(TEST_UTIL.getConfiguration(), tableName, family);
151     FileStatus[] fileStatuses = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
152     // mobFileSet0 stores the orignal mob files
153     TreeSet<String> mobFilesSet = new TreeSet<String>();
154     for (FileStatus status : fileStatuses) {
155       mobFilesSet.add(status.getPath().getName());
156     }
157 
158     //scan the table, retreive the references
159     Scan scan = new Scan();
160     scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
161     scan.setAttribute(MobConstants.MOB_SCAN_REF_ONLY, Bytes.toBytes(Boolean.TRUE));
162     ResultScanner rs = table.getScanner(scan);
163     TreeSet<String> mobFilesScanned = new TreeSet<String>();
164     for (Result res : rs) {
165       byte[] valueBytes = res.getValue(Bytes.toBytes(family),
166           Bytes.toBytes(column));
167       mobFilesScanned.add(Bytes.toString(valueBytes, Bytes.SIZEOF_INT,
168           valueBytes.length - Bytes.SIZEOF_INT));
169     }
170     //there should be 10 mob files
171     assertEquals(10, mobFilesScanned.size());
172     //check if we store the correct reference of mob files
173     assertEquals(mergeString(mobFilesSet), mergeString(mobFilesScanned));
174 
175     Configuration conf = TEST_UTIL.getConfiguration();
176     conf.setLong(SweepJob.MOB_SWEEP_JOB_DELAY, 24 * 60 * 60 * 1000);
177 
178     String[] args = new String[2];
179     args[0] = tableName;
180     args[1] = family;
181     assertEquals(0, ToolRunner.run(conf, new Sweeper(), args));
182 
183     mobFamilyPath = getMobFamilyPath(TEST_UTIL.getConfiguration(), tableName, family);
184     fileStatuses = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
185     mobFilesSet = new TreeSet<String>();
186     for (FileStatus status : fileStatuses) {
187       mobFilesSet.add(status.getPath().getName());
188     }
189     assertEquals(10, mobFilesSet.size());
190 
191     scan = new Scan();
192     scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
193     scan.setAttribute(MobConstants.MOB_SCAN_REF_ONLY, Bytes.toBytes(Boolean.TRUE));
194     rs = table.getScanner(scan);
195     TreeSet<String> mobFilesScannedAfterJob = new TreeSet<String>();
196     for (Result res : rs) {
197       byte[] valueBytes = res.getValue(Bytes.toBytes(family), Bytes.toBytes(
198           column));
199       mobFilesScannedAfterJob.add(Bytes.toString(valueBytes, Bytes.SIZEOF_INT,
200           valueBytes.length - Bytes.SIZEOF_INT));
201     }
202     assertEquals(10, mobFilesScannedAfterJob.size());
203 
204     fileStatuses = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
205     mobFilesSet = new TreeSet<String>();
206     for (FileStatus status : fileStatuses) {
207       mobFilesSet.add(status.getPath().getName());
208     }
209     assertEquals(10, mobFilesSet.size());
210     assertEquals(true, mobFilesScannedAfterJob.iterator().next()
211             .equalsIgnoreCase(mobFilesSet.iterator().next()));
212   }
213 
214   private void testCompactionDelaySweeperInternal(HTable table, String tableName)
215     throws Exception {
216     int count = 10;
217     //create table and generate 10 mob files
218     generateMobTable(admin, table, tableName, count, 1);
219     //get mob files
220     Path mobFamilyPath = getMobFamilyPath(TEST_UTIL.getConfiguration(), tableName, family);
221     FileStatus[] fileStatuses = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
222     // mobFileSet0 stores the orignal mob files
223     TreeSet<String> mobFilesSet = new TreeSet<String>();
224     for (FileStatus status : fileStatuses) {
225       mobFilesSet.add(status.getPath().getName());
226     }
227 
228     //scan the table, retreive the references
229     Scan scan = new Scan();
230     scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
231     scan.setAttribute(MobConstants.MOB_SCAN_REF_ONLY, Bytes.toBytes(Boolean.TRUE));
232     ResultScanner rs = table.getScanner(scan);
233     TreeSet<String> mobFilesScanned = new TreeSet<String>();
234     for (Result res : rs) {
235       byte[] valueBytes = res.getValue(Bytes.toBytes(family),
236               Bytes.toBytes(column));
237       mobFilesScanned.add(Bytes.toString(valueBytes, Bytes.SIZEOF_INT,
238           valueBytes.length - Bytes.SIZEOF_INT));
239     }
240     //there should be 10 mob files
241     assertEquals(10, mobFilesScanned.size());
242     //check if we store the correct reference of mob files
243     assertEquals(mergeString(mobFilesSet), mergeString(mobFilesScanned));
244 
245     Configuration conf = TEST_UTIL.getConfiguration();
246     conf.setLong(SweepJob.MOB_SWEEP_JOB_DELAY, 0);
247     String[] args = new String[2];
248     args[0] = tableName;
249     args[1] = family;
250     assertEquals(0, ToolRunner.run(conf, new Sweeper(), args));
251 
252     mobFamilyPath = getMobFamilyPath(TEST_UTIL.getConfiguration(), tableName, family);
253     fileStatuses = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
254     mobFilesSet = new TreeSet<String>();
255     for (FileStatus status : fileStatuses) {
256       mobFilesSet.add(status.getPath().getName());
257     }
258     assertEquals(1, mobFilesSet.size());
259 
260     scan = new Scan();
261     scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
262     scan.setAttribute(MobConstants.MOB_SCAN_REF_ONLY, Bytes.toBytes(Boolean.TRUE));
263     rs = table.getScanner(scan);
264     TreeSet<String> mobFilesScannedAfterJob = new TreeSet<String>();
265     for (Result res : rs) {
266       byte[] valueBytes = res.getValue(Bytes.toBytes(family), Bytes.toBytes(
267               column));
268       mobFilesScannedAfterJob.add(Bytes.toString(valueBytes, Bytes.SIZEOF_INT,
269           valueBytes.length - Bytes.SIZEOF_INT));
270     }
271     assertEquals(1, mobFilesScannedAfterJob.size());
272 
273     fileStatuses = TEST_UTIL.getTestFileSystem().listStatus(mobFamilyPath);
274     mobFilesSet = new TreeSet<String>();
275     for (FileStatus status : fileStatuses) {
276       mobFilesSet.add(status.getPath().getName());
277     }
278     assertEquals(1, mobFilesSet.size());
279     assertEquals(true, mobFilesScannedAfterJob.iterator().next()
280             .equalsIgnoreCase(mobFilesSet.iterator().next()));
281   }
282 
283   @Test
284   public void testCompactionDelaySweeper() throws Exception {
285     testCompactionDelaySweeperInternal(table, tableName);
286   }
287 
288   @Test
289   public void testCompactionDelaySweeperWithNamespace() throws Exception {
290     // create a table with namespace
291     NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create("ns").build();
292     admin.createNamespace(namespaceDescriptor);
293     String tableNameAsString = "ns:testSweeperWithNamespace";
294     TableName tableName = TableName.valueOf(tableNameAsString);
295     HTableDescriptor desc = new HTableDescriptor(tableName);
296     HColumnDescriptor hcd = new HColumnDescriptor(family);
297     hcd.setMobEnabled(true);
298     hcd.setMobThreshold(3L);
299     hcd.setMaxVersions(4);
300     desc.addFamily(hcd);
301     admin.createTable(desc);
302     HTable table = new HTable(TEST_UTIL.getConfiguration(), tableName);
303     table.setAutoFlush(false, false);
304     testCompactionDelaySweeperInternal(table, tableNameAsString);
305     table.close();
306     admin.disableTable(tableName);
307     admin.deleteTable(tableName);
308     admin.deleteNamespace("ns");
309   }
310 }