hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestJoinedScanners.java

   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.regionserver;
  19
  20 import java.io.IOException;
  21 import java.util.ArrayList;
  22 import java.util.List;
  23 import java.util.Random;
  24 import org.apache.hadoop.hbase.CompareOperator;
  25 import org.apache.hadoop.hbase.DoNotRetryIOException;
  26 import org.apache.hadoop.hbase.HBaseClassTestRule;
  27 import org.apache.hadoop.hbase.HBaseTestingUtil;
  28 import org.apache.hadoop.hbase.StartTestingClusterOption;
  29 import org.apache.hadoop.hbase.TableName;
  30 import org.apache.hadoop.hbase.client.Admin;
  31 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
  32 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
  33 import org.apache.hadoop.hbase.client.Connection;
  34 import org.apache.hadoop.hbase.client.Put;
  35 import org.apache.hadoop.hbase.client.Result;
  36 import org.apache.hadoop.hbase.client.ResultScanner;
  37 import org.apache.hadoop.hbase.client.Scan;
  38 import org.apache.hadoop.hbase.client.Table;
  39 import org.apache.hadoop.hbase.client.TableDescriptor;
  40 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
  41 import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
  42 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
  43 import org.apache.hadoop.hbase.testclassification.LargeTests;
  44 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
  45 import org.apache.hadoop.hbase.util.Bytes;
  46 import org.junit.AfterClass;
  47 import org.junit.BeforeClass;
  48 import org.junit.ClassRule;
  49 import org.junit.Rule;
  50 import org.junit.Test;
  51 import org.junit.experimental.categories.Category;
  52 import org.junit.rules.TestName;
  53 import org.slf4j.Logger;
  54 import org.slf4j.LoggerFactory;
  55
  56 import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
  57 import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLineParser;
  58 import org.apache.hbase.thirdparty.org.apache.commons.cli.GnuParser;
  59 import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter;
  60 import org.apache.hbase.thirdparty.org.apache.commons.cli.Option;
  61 import org.apache.hbase.thirdparty.org.apache.commons.cli.Options;
  62
  63 /**
  64  * Test performance improvement of joined scanners optimization:
  65  * https://issues.apache.org/jira/browse/HBASE-5416
  66  */
  67 @Category({RegionServerTests.class, LargeTests.class})
  68 public class TestJoinedScanners {
  69
  70   @ClassRule
  71   public static final HBaseClassTestRule CLASS_RULE =
  72       HBaseClassTestRule.forClass(TestJoinedScanners.class);
  73
  74   private static final Logger LOG = LoggerFactory.getLogger(TestJoinedScanners.class);
  75
  76   private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
  77
  78   private static final byte[] cf_essential = Bytes.toBytes("essential");
  79   private static final byte[] cf_joined = Bytes.toBytes("joined");
  80   private static final byte[] col_name = Bytes.toBytes("a");
  81   private static final byte[] flag_yes = Bytes.toBytes("Y");
  82   private static final byte[] flag_no = Bytes.toBytes("N");
  83
  84   private static DataBlockEncoding blockEncoding = DataBlockEncoding.FAST_DIFF;
  85   private static int selectionRatio = 30;
  86   private static int valueWidth = 128 * 1024;
  87
  88   @Rule
  89   public TestName name = new TestName();
  90
  91   @BeforeClass
  92   public static void setUpBeforeClass() throws Exception {
  93     final int DEFAULT_BLOCK_SIZE = 1024 * 1024;
  94     TEST_UTIL.getConfiguration().setLong("dfs.blocksize", DEFAULT_BLOCK_SIZE);
  95     TEST_UTIL.getConfiguration().setInt("dfs.replication", 1);
  96     TEST_UTIL.getConfiguration().setLong("hbase.hregion.max.filesize", 322122547200L);
  97
  98     String[] dataNodeHosts = new String[] {"host1", "host2", "host3"};
  99     int regionServersCount = 3;
 100     StartTestingClusterOption option = StartTestingClusterOption.builder()
 101         .numRegionServers(regionServersCount).dataNodeHosts(dataNodeHosts).build();
 102     TEST_UTIL.startMiniCluster(option);
 103   }
 104
 105   @AfterClass
 106   public static void tearDownAfterClass() throws Exception {
 107     TEST_UTIL.shutdownMiniCluster();
 108   }
 109
 110   @Test
 111   public void testJoinedScanners() throws Exception {
 112     byte[][] families = {cf_essential, cf_joined};
 113
 114     final TableName tableName = TableName.valueOf(name.getMethodName());
 115     TableDescriptorBuilder builder =
 116       TableDescriptorBuilder.newBuilder(tableName);
 117     for (byte[] family : families) {
 118       ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(family)
 119         .setDataBlockEncoding(blockEncoding).build();
 120       builder.setColumnFamily(familyDescriptor);
 121     }
 122     TableDescriptor tableDescriptor = builder.build();
 123     TEST_UTIL.getAdmin().createTable(tableDescriptor);
 124     Table ht = TEST_UTIL.getConnection().getTable(tableName);
 125
 126     long rows_to_insert = 1000;
 127     int insert_batch = 20;
 128     long time = System.nanoTime();
 129     Random rand = new Random(time);
 130
 131     LOG.info("Make " + Long.toString(rows_to_insert) + " rows, total size = " + Float
 132       .toString(rows_to_insert * valueWidth / 1024 / 1024) + " MB");
 133
 134     byte[] val_large = new byte[valueWidth];
 135
 136     List<Put> puts = new ArrayList<>();
 137
 138     for (long i = 0; i < rows_to_insert; i++) {
 139       Put put = new Put(Bytes.toBytes(Long.toString(i)));
 140       if (rand.nextInt(100) <= selectionRatio) {
 141         put.addColumn(cf_essential, col_name, flag_yes);
 142       } else {
 143         put.addColumn(cf_essential, col_name, flag_no);
 144       }
 145       put.addColumn(cf_joined, col_name, val_large);
 146       puts.add(put);
 147       if (puts.size() >= insert_batch) {
 148         ht.put(puts);
 149         puts.clear();
 150       }
 151     }
 152     if (!puts.isEmpty()) {
 153       ht.put(puts);
 154       puts.clear();
 155     }
 156
 157     LOG.info("Data generated in "
 158       + Double.toString((System.nanoTime() - time) / 1000000000.0) + " seconds");
 159
 160     boolean slow = true;
 161     for (int i = 0; i < 10; ++i) {
 162       runScanner(ht, slow);
 163       slow = !slow;
 164     }
 165
 166     ht.close();
 167   }
 168
 169   private void runScanner(Table table, boolean slow) throws Exception {
 170     long time = System.nanoTime();
 171     Scan scan = new Scan();
 172     scan.addColumn(cf_essential, col_name);
 173     scan.addColumn(cf_joined, col_name);
 174
 175     SingleColumnValueFilter filter = new SingleColumnValueFilter(
 176         cf_essential, col_name, CompareOperator.EQUAL, flag_yes);
 177     filter.setFilterIfMissing(true);
 178     scan.setFilter(filter);
 179     scan.setLoadColumnFamiliesOnDemand(!slow);
 180
 181     ResultScanner result_scanner = table.getScanner(scan);
 182     Result res;
 183     long rows_count = 0;
 184     while ((res = result_scanner.next()) != null) {
 185       rows_count++;
 186     }
 187
 188     double timeSec = (System.nanoTime() - time) / 1000000000.0;
 189     result_scanner.close();
 190     LOG.info((slow ? "Slow" : "Joined") + " scanner finished in " + Double.toString(timeSec)
 191       + " seconds, got " + Long.toString(rows_count/2) + " rows");
 192   }
 193
 194   private static Options options = new Options();
 195
 196   /**
 197    * Command line interface:
 198    * @param args
 199    * @throws IOException if there is a bug while reading from disk
 200    */
 201   public static void main(final String[] args) throws Exception {
 202     Option encodingOption = new Option("e", "blockEncoding", true,
 203       "Data block encoding; Default: FAST_DIFF");
 204     encodingOption.setRequired(false);
 205     options.addOption(encodingOption);
 206
 207     Option ratioOption = new Option("r", "selectionRatio", true,
 208       "Ratio of selected rows using essential column family");
 209     ratioOption.setRequired(false);
 210     options.addOption(ratioOption);
 211
 212     Option widthOption = new Option("w", "valueWidth", true,
 213       "Width of value for non-essential column family");
 214     widthOption.setRequired(false);
 215     options.addOption(widthOption);
 216
 217     CommandLineParser parser = new GnuParser();
 218     CommandLine cmd = parser.parse(options, args);
 219     if (args.length < 1) {
 220       HelpFormatter formatter = new HelpFormatter();
 221       formatter.printHelp("TestJoinedScanners", options, true);
 222     }
 223
 224     if (cmd.hasOption("e")) {
 225       blockEncoding = DataBlockEncoding.valueOf(cmd.getOptionValue("e"));
 226     }
 227     if (cmd.hasOption("r")) {
 228       selectionRatio = Integer.parseInt(cmd.getOptionValue("r"));
 229     }
 230     if (cmd.hasOption("w")) {
 231       valueWidth = Integer.parseInt(cmd.getOptionValue("w"));
 232     }
 233     // run the test
 234     TestJoinedScanners test = new TestJoinedScanners();
 235     test.testJoinedScanners();
 236   }
 237
 238   @Test(expected = DoNotRetryIOException.class)
 239   public void testWithReverseScan() throws Exception {
 240     try (Connection con = TEST_UTIL.getConnection(); Admin admin = con.getAdmin()) {
 241       TableName tableName = TableName.valueOf(name.getMethodName());
 242
 243       TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
 244           .setColumnFamily(ColumnFamilyDescriptorBuilder.of("cf1"))
 245           .setColumnFamily(ColumnFamilyDescriptorBuilder.of("cf2"))
 246           .build();
 247       admin.createTable(tableDescriptor);
 248
 249       try (Table table = con.getTable(tableName)) {
 250         SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("cf1"),
 251           Bytes.toBytes("col"), CompareOperator.EQUAL, Bytes.toBytes("val"));
 252         filter.setFilterIfMissing(true);
 253
 254         // Reverse scan with loading CFs on demand
 255         Scan scan = new Scan();
 256         scan.setFilter(filter);
 257         scan.setReversed(true);
 258         scan.setLoadColumnFamiliesOnDemand(true);
 259
 260         try (ResultScanner scanner = table.getScanner(scan)) {
 261           // DoNotRetryIOException should occur
 262           scanner.next();
 263         }
 264       }
 265     }
 266   }
 267 }