001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.regionserver; 019 020import java.io.IOException; 021import java.util.ArrayList; 022import java.util.List; 023import java.util.Random; 024 025import org.apache.hadoop.hbase.CompareOperator; 026import org.apache.hadoop.hbase.DoNotRetryIOException; 027import org.apache.hadoop.hbase.HBaseClassTestRule; 028import org.apache.hadoop.hbase.HBaseTestingUtility; 029import org.apache.hadoop.hbase.HColumnDescriptor; 030import org.apache.hadoop.hbase.HTableDescriptor; 031import org.apache.hadoop.hbase.TableName; 032import org.apache.hadoop.hbase.client.Admin; 033import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; 034import org.apache.hadoop.hbase.client.Connection; 035import org.apache.hadoop.hbase.client.Put; 036import org.apache.hadoop.hbase.client.Result; 037import org.apache.hadoop.hbase.client.ResultScanner; 038import org.apache.hadoop.hbase.client.Scan; 039import org.apache.hadoop.hbase.client.Table; 040import org.apache.hadoop.hbase.client.TableDescriptor; 041import org.apache.hadoop.hbase.client.TableDescriptorBuilder; 042import org.apache.hadoop.hbase.filter.CompareFilter; 043import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; 044import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding; 045import org.apache.hadoop.hbase.testclassification.LargeTests; 046import org.apache.hadoop.hbase.testclassification.RegionServerTests; 047import org.apache.hadoop.hbase.util.Bytes; 048import org.junit.AfterClass; 049import org.junit.BeforeClass; 050import org.junit.ClassRule; 051import org.junit.Rule; 052import org.junit.Test; 053import org.junit.experimental.categories.Category; 054import org.junit.rules.TestName; 055import org.slf4j.Logger; 056import org.slf4j.LoggerFactory; 057import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine; 058import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLineParser; 059import org.apache.hbase.thirdparty.org.apache.commons.cli.GnuParser; 060import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter; 061import org.apache.hbase.thirdparty.org.apache.commons.cli.Option; 062import org.apache.hbase.thirdparty.org.apache.commons.cli.Options; 063 064/** 065 * Test performance improvement of joined scanners optimization: 066 * https://issues.apache.org/jira/browse/HBASE-5416 067 */ 068@Category({RegionServerTests.class, LargeTests.class}) 069public class TestJoinedScanners { 070 071 @ClassRule 072 public static final HBaseClassTestRule CLASS_RULE = 073 HBaseClassTestRule.forClass(TestJoinedScanners.class); 074 075 private static final Logger LOG = LoggerFactory.getLogger(TestJoinedScanners.class); 076 077 private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); 078 079 private static final byte[] cf_essential = Bytes.toBytes("essential"); 080 private static final byte[] cf_joined = Bytes.toBytes("joined"); 081 private static final byte[] col_name = Bytes.toBytes("a"); 082 private static final byte[] flag_yes = Bytes.toBytes("Y"); 083 private static final byte[] flag_no = Bytes.toBytes("N"); 084 085 private static DataBlockEncoding blockEncoding = DataBlockEncoding.FAST_DIFF; 086 private static int selectionRatio = 30; 087 private static int valueWidth = 128 * 1024; 088 089 @Rule 090 public TestName name = new TestName(); 091 092 @BeforeClass 093 public static void setUpBeforeClass() throws Exception { 094 final int DEFAULT_BLOCK_SIZE = 1024 * 1024; 095 TEST_UTIL.getConfiguration().setLong("dfs.blocksize", DEFAULT_BLOCK_SIZE); 096 TEST_UTIL.getConfiguration().setInt("dfs.replication", 1); 097 TEST_UTIL.getConfiguration().setLong("hbase.hregion.max.filesize", 322122547200L); 098 099 String[] dataNodeHosts = new String[] {"host1", "host2", "host3"}; 100 int regionServersCount = 3; 101 TEST_UTIL.startMiniCluster(1, regionServersCount, dataNodeHosts); 102 } 103 104 @AfterClass 105 public static void tearDownAfterClass() throws Exception { 106 TEST_UTIL.shutdownMiniCluster(); 107 } 108 109 @Test 110 public void testJoinedScanners() throws Exception { 111 byte[][] families = {cf_essential, cf_joined}; 112 113 final TableName tableName = TableName.valueOf(name.getMethodName()); 114 HTableDescriptor desc = new HTableDescriptor(tableName); 115 for (byte[] family : families) { 116 HColumnDescriptor hcd = new HColumnDescriptor(family); 117 hcd.setDataBlockEncoding(blockEncoding); 118 desc.addFamily(hcd); 119 } 120 TEST_UTIL.getAdmin().createTable(desc); 121 Table ht = TEST_UTIL.getConnection().getTable(tableName); 122 123 long rows_to_insert = 1000; 124 int insert_batch = 20; 125 long time = System.nanoTime(); 126 Random rand = new Random(time); 127 128 LOG.info("Make " + Long.toString(rows_to_insert) + " rows, total size = " + Float 129 .toString(rows_to_insert * valueWidth / 1024 / 1024) + " MB"); 130 131 byte[] val_large = new byte[valueWidth]; 132 133 List<Put> puts = new ArrayList<>(); 134 135 for (long i = 0; i < rows_to_insert; i++) { 136 Put put = new Put(Bytes.toBytes(Long.toString(i))); 137 if (rand.nextInt(100) <= selectionRatio) { 138 put.addColumn(cf_essential, col_name, flag_yes); 139 } else { 140 put.addColumn(cf_essential, col_name, flag_no); 141 } 142 put.addColumn(cf_joined, col_name, val_large); 143 puts.add(put); 144 if (puts.size() >= insert_batch) { 145 ht.put(puts); 146 puts.clear(); 147 } 148 } 149 if (!puts.isEmpty()) { 150 ht.put(puts); 151 puts.clear(); 152 } 153 154 LOG.info("Data generated in " 155 + Double.toString((System.nanoTime() - time) / 1000000000.0) + " seconds"); 156 157 boolean slow = true; 158 for (int i = 0; i < 10; ++i) { 159 runScanner(ht, slow); 160 slow = !slow; 161 } 162 163 ht.close(); 164 } 165 166 private void runScanner(Table table, boolean slow) throws Exception { 167 long time = System.nanoTime(); 168 Scan scan = new Scan(); 169 scan.addColumn(cf_essential, col_name); 170 scan.addColumn(cf_joined, col_name); 171 172 SingleColumnValueFilter filter = new SingleColumnValueFilter( 173 cf_essential, col_name, CompareFilter.CompareOp.EQUAL, flag_yes); 174 filter.setFilterIfMissing(true); 175 scan.setFilter(filter); 176 scan.setLoadColumnFamiliesOnDemand(!slow); 177 178 ResultScanner result_scanner = table.getScanner(scan); 179 Result res; 180 long rows_count = 0; 181 while ((res = result_scanner.next()) != null) { 182 rows_count++; 183 } 184 185 double timeSec = (System.nanoTime() - time) / 1000000000.0; 186 result_scanner.close(); 187 LOG.info((slow ? "Slow" : "Joined") + " scanner finished in " + Double.toString(timeSec) 188 + " seconds, got " + Long.toString(rows_count/2) + " rows"); 189 } 190 191 private static Options options = new Options(); 192 193 /** 194 * Command line interface: 195 * @param args 196 * @throws IOException if there is a bug while reading from disk 197 */ 198 public static void main(final String[] args) throws Exception { 199 Option encodingOption = new Option("e", "blockEncoding", true, 200 "Data block encoding; Default: FAST_DIFF"); 201 encodingOption.setRequired(false); 202 options.addOption(encodingOption); 203 204 Option ratioOption = new Option("r", "selectionRatio", true, 205 "Ratio of selected rows using essential column family"); 206 ratioOption.setRequired(false); 207 options.addOption(ratioOption); 208 209 Option widthOption = new Option("w", "valueWidth", true, 210 "Width of value for non-essential column family"); 211 widthOption.setRequired(false); 212 options.addOption(widthOption); 213 214 CommandLineParser parser = new GnuParser(); 215 CommandLine cmd = parser.parse(options, args); 216 if (args.length < 1) { 217 HelpFormatter formatter = new HelpFormatter(); 218 formatter.printHelp("TestJoinedScanners", options, true); 219 } 220 221 if (cmd.hasOption("e")) { 222 blockEncoding = DataBlockEncoding.valueOf(cmd.getOptionValue("e")); 223 } 224 if (cmd.hasOption("r")) { 225 selectionRatio = Integer.parseInt(cmd.getOptionValue("r")); 226 } 227 if (cmd.hasOption("w")) { 228 valueWidth = Integer.parseInt(cmd.getOptionValue("w")); 229 } 230 // run the test 231 TestJoinedScanners test = new TestJoinedScanners(); 232 test.testJoinedScanners(); 233 } 234 235 @Test(expected = DoNotRetryIOException.class) 236 public void testWithReverseScan() throws Exception { 237 try (Connection con = TEST_UTIL.getConnection(); Admin admin = con.getAdmin()) { 238 TableName tableName = TableName.valueOf(name.getMethodName()); 239 240 TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName) 241 .setColumnFamily(ColumnFamilyDescriptorBuilder.of("cf1")) 242 .setColumnFamily(ColumnFamilyDescriptorBuilder.of("cf2")) 243 .build(); 244 admin.createTable(tableDescriptor); 245 246 try (Table table = con.getTable(tableName)) { 247 SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("cf1"), 248 Bytes.toBytes("col"), CompareOperator.EQUAL, Bytes.toBytes("val")); 249 filter.setFilterIfMissing(true); 250 251 // Reverse scan with loading CFs on demand 252 Scan scan = new Scan(); 253 scan.setFilter(filter); 254 scan.setReversed(true); 255 scan.setLoadColumnFamiliesOnDemand(true); 256 257 try (ResultScanner scanner = table.getScanner(scan)) { 258 // DoNotRetryIOException should occur 259 scanner.next(); 260 } 261 } 262 } 263 } 264}