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; 019 020import static org.junit.Assert.assertEquals; 021import static org.junit.Assert.assertNotNull; 022import static org.junit.Assert.assertTrue; 023 024import java.io.IOException; 025import java.util.ArrayList; 026import java.util.List; 027import org.apache.hadoop.hbase.client.Put; 028import org.apache.hadoop.hbase.client.Result; 029import org.apache.hadoop.hbase.client.ResultScanner; 030import org.apache.hadoop.hbase.client.Scan; 031import org.apache.hadoop.hbase.client.Table; 032import org.apache.hadoop.hbase.client.metrics.ScanMetrics; 033import org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics; 034import org.apache.hadoop.hbase.filter.BinaryComparator; 035import org.apache.hadoop.hbase.filter.ColumnPrefixFilter; 036import org.apache.hadoop.hbase.filter.Filter; 037import org.apache.hadoop.hbase.filter.FilterList; 038import org.apache.hadoop.hbase.filter.FilterList.Operator; 039import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter; 040import org.apache.hadoop.hbase.filter.RowFilter; 041import org.apache.hadoop.hbase.filter.SingleColumnValueExcludeFilter; 042import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; 043import org.apache.hadoop.hbase.testclassification.LargeTests; 044import org.apache.hadoop.hbase.util.Bytes; 045import org.junit.AfterClass; 046import org.junit.BeforeClass; 047import org.junit.ClassRule; 048import org.junit.Test; 049import org.junit.experimental.categories.Category; 050import org.slf4j.Logger; 051import org.slf4j.LoggerFactory; 052 053@Category(LargeTests.class) 054public class TestServerSideScanMetricsFromClientSide { 055 private static final Logger LOG = 056 LoggerFactory.getLogger(TestServerSideScanMetricsFromClientSide.class); 057 058 @ClassRule 059 public static final HBaseClassTestRule CLASS_RULE = 060 HBaseClassTestRule.forClass(TestServerSideScanMetricsFromClientSide.class); 061 062 private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 063 064 private static Table TABLE = null; 065 066 /** 067 * Table configuration 068 */ 069 private static TableName TABLE_NAME = TableName.valueOf("testTable"); 070 071 private static int NUM_ROWS = 10; 072 private static byte[] ROW = Bytes.toBytes("testRow"); 073 private static byte[][] ROWS = HTestConst.makeNAscii(ROW, NUM_ROWS); 074 075 // Should keep this value below 10 to keep generation of expected kv's simple. If above 10 then 076 // table/row/cf1/... will be followed by table/row/cf10/... instead of table/row/cf2/... which 077 // breaks the simple generation of expected kv's 078 private static int NUM_FAMILIES = 1; 079 private static byte[] FAMILY = Bytes.toBytes("testFamily"); 080 private static byte[][] FAMILIES = HTestConst.makeNAscii(FAMILY, NUM_FAMILIES); 081 082 private static int NUM_QUALIFIERS = 1; 083 private static byte[] QUALIFIER = Bytes.toBytes("testQualifier"); 084 private static byte[][] QUALIFIERS = HTestConst.makeNAscii(QUALIFIER, NUM_QUALIFIERS); 085 086 private static int VALUE_SIZE = 10; 087 private static byte[] VALUE = Bytes.createMaxByteArray(VALUE_SIZE); 088 089 private static int NUM_COLS = NUM_FAMILIES * NUM_QUALIFIERS; 090 091 // Approximation of how large the heap size of cells in our table. Should be accessed through 092 // getCellHeapSize(). 093 private static long CELL_HEAP_SIZE = -1; 094 095 @BeforeClass 096 public static void setUpBeforeClass() throws Exception { 097 TEST_UTIL.startMiniCluster(3); 098 TABLE = createTestTable(TABLE_NAME, ROWS, FAMILIES, QUALIFIERS, VALUE); 099 } 100 101 private static Table createTestTable(TableName name, byte[][] rows, byte[][] families, 102 byte[][] qualifiers, byte[] cellValue) throws IOException { 103 Table ht = TEST_UTIL.createTable(name, families); 104 List<Put> puts = createPuts(rows, families, qualifiers, cellValue); 105 ht.put(puts); 106 107 return ht; 108 } 109 110 @AfterClass 111 public static void tearDownAfterClass() throws Exception { 112 TEST_UTIL.shutdownMiniCluster(); 113 } 114 115 /** 116 * Make puts to put the input value into each combination of row, family, and qualifier 117 * @param rows the rows to use 118 * @param families the column families to use 119 * @param qualifiers the column qualifiers to use 120 * @param value the value to put 121 * @return the putted input values added in puts 122 * @throws IOException If an IO problem is encountered 123 */ 124 private static ArrayList<Put> createPuts(byte[][] rows, byte[][] families, byte[][] qualifiers, 125 byte[] value) throws IOException { 126 Put put; 127 ArrayList<Put> puts = new ArrayList<>(); 128 129 for (int row = 0; row < rows.length; row++) { 130 put = new Put(rows[row]); 131 for (int fam = 0; fam < families.length; fam++) { 132 for (int qual = 0; qual < qualifiers.length; qual++) { 133 KeyValue kv = new KeyValue(rows[row], families[fam], qualifiers[qual], qual, value); 134 put.add(kv); 135 } 136 } 137 puts.add(put); 138 } 139 140 return puts; 141 } 142 143 /** 144 * @return The approximate heap size of a cell in the test table. All cells should have 145 * approximately the same heap size, so the value is cached to avoid repeating the 146 * calculation 147 * @throws Exception on unexpected failure 148 */ 149 private long getCellHeapSize() throws Exception { 150 if (CELL_HEAP_SIZE == -1) { 151 // Do a partial scan that will return a single result with a single cell 152 Scan scan = new Scan(); 153 scan.setMaxResultSize(1); 154 scan.setAllowPartialResults(true); 155 ResultScanner scanner = TABLE.getScanner(scan); 156 157 Result result = scanner.next(); 158 159 assertTrue(result != null); 160 assertTrue(result.rawCells() != null); 161 assertTrue(result.rawCells().length == 1); 162 163 CELL_HEAP_SIZE = result.rawCells()[0].heapSize(); 164 scanner.close(); 165 } 166 167 return CELL_HEAP_SIZE; 168 } 169 170 @Test 171 public void testRowsSeenMetric() throws Exception { 172 // Base scan configuration 173 Scan baseScan; 174 baseScan = new Scan(); 175 baseScan.setScanMetricsEnabled(true); 176 try { 177 testRowsSeenMetric(baseScan); 178 179 // Test case that only a single result will be returned per RPC to the serer 180 baseScan.setCaching(1); 181 testRowsSeenMetric(baseScan); 182 183 // Test case that partial results are returned from the server. At most one cell will be 184 // contained in each response 185 baseScan.setMaxResultSize(1); 186 testRowsSeenMetric(baseScan); 187 188 // Test case that size limit is set such that a few cells are returned per partial result from 189 // the server 190 baseScan.setCaching(NUM_ROWS); 191 baseScan.setMaxResultSize(getCellHeapSize() * (NUM_COLS - 1)); 192 testRowsSeenMetric(baseScan); 193 } catch (Throwable t) { 194 LOG.error("FAIL", t); 195 throw t; 196 } 197 } 198 199 @Test 200 public void testFsReadTimeMetric() throws Exception { 201 // write some new puts and flush, as an easy way to ensure the read blocks are not cached 202 // so that we go into the fs write code path 203 List<Put> puts = createPuts(ROWS, FAMILIES, QUALIFIERS, VALUE); 204 TABLE.put(puts); 205 TEST_UTIL.flush(TABLE_NAME); 206 Scan scan = new Scan(); 207 scan.setScanMetricsEnabled(true); 208 testMetric(scan, ServerSideScanMetrics.FS_READ_TIME_METRIC_NAME, 0, CompareOperator.GREATER); 209 } 210 211 private void testRowsSeenMetric(Scan baseScan) throws Exception { 212 Scan scan; 213 scan = new Scan(baseScan); 214 testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, NUM_ROWS); 215 216 for (int i = 0; i < ROWS.length - 1; i++) { 217 scan = new Scan(baseScan); 218 scan.withStartRow(ROWS[0]); 219 scan.withStopRow(ROWS[i + 1]); 220 testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, i + 1); 221 } 222 223 for (int i = ROWS.length - 1; i > 0; i--) { 224 scan = new Scan(baseScan); 225 scan.withStartRow(ROWS[i - 1]); 226 scan.withStopRow(ROWS[ROWS.length - 1]); 227 testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, 228 ROWS.length - i); 229 } 230 231 // The filter should filter out all rows, but we still expect to see every row. 232 Filter filter = 233 new RowFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("xyz"))); 234 scan = new Scan(baseScan); 235 scan.setFilter(filter); 236 testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, ROWS.length); 237 238 // Filter should pass on all rows 239 SingleColumnValueFilter singleColumnValueFilter = 240 new SingleColumnValueFilter(FAMILIES[0], QUALIFIERS[0], CompareOperator.EQUAL, VALUE); 241 scan = new Scan(baseScan); 242 scan.setFilter(singleColumnValueFilter); 243 testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, ROWS.length); 244 245 // Filter should filter out all rows 246 singleColumnValueFilter = 247 new SingleColumnValueFilter(FAMILIES[0], QUALIFIERS[0], CompareOperator.NOT_EQUAL, VALUE); 248 scan = new Scan(baseScan); 249 scan.setFilter(singleColumnValueFilter); 250 testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME, ROWS.length); 251 } 252 253 @Test 254 public void testRowsFilteredMetric() throws Exception { 255 // Base scan configuration 256 Scan baseScan; 257 baseScan = new Scan(); 258 baseScan.setScanMetricsEnabled(true); 259 260 // Test case where scan uses default values 261 testRowsFilteredMetric(baseScan); 262 263 // Test case where at most one Result is retrieved per RPC 264 baseScan.setCaching(1); 265 testRowsFilteredMetric(baseScan); 266 267 // Test case where size limit is very restrictive and partial results will be returned from 268 // server 269 baseScan.setMaxResultSize(1); 270 testRowsFilteredMetric(baseScan); 271 272 // Test a case where max result size limits response from server to only a few cells (not all 273 // cells from the row) 274 baseScan.setCaching(NUM_ROWS); 275 baseScan.setMaxResultSize(getCellHeapSize() * (NUM_COLS - 1)); 276 testRowsSeenMetric(baseScan); 277 } 278 279 private void testRowsFilteredMetric(Scan baseScan) throws Exception { 280 testRowsFilteredMetric(baseScan, null, 0); 281 282 // Row filter doesn't match any row key. All rows should be filtered 283 Filter filter = 284 new RowFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("xyz"))); 285 testRowsFilteredMetric(baseScan, filter, ROWS.length); 286 287 // Filter will return results containing only the first key. Number of entire rows filtered 288 // should be 0. 289 filter = new FirstKeyOnlyFilter(); 290 testRowsFilteredMetric(baseScan, filter, 0); 291 292 // Column prefix will find some matching qualifier on each row. Number of entire rows filtered 293 // should be 0 294 filter = new ColumnPrefixFilter(QUALIFIERS[0]); 295 testRowsFilteredMetric(baseScan, filter, 0); 296 297 // Column prefix will NOT find any matching qualifier on any row. All rows should be filtered 298 filter = new ColumnPrefixFilter(Bytes.toBytes("xyz")); 299 testRowsFilteredMetric(baseScan, filter, ROWS.length); 300 301 // Matching column value should exist in each row. No rows should be filtered. 302 filter = new SingleColumnValueFilter(FAMILIES[0], QUALIFIERS[0], CompareOperator.EQUAL, VALUE); 303 testRowsFilteredMetric(baseScan, filter, 0); 304 305 // No matching column value should exist in any row. Filter all rows 306 filter = 307 new SingleColumnValueFilter(FAMILIES[0], QUALIFIERS[0], CompareOperator.NOT_EQUAL, VALUE); 308 testRowsFilteredMetric(baseScan, filter, ROWS.length); 309 310 List<Filter> filters = new ArrayList<>(); 311 filters.add(new RowFilter(CompareOperator.EQUAL, new BinaryComparator(ROWS[0]))); 312 filters.add(new RowFilter(CompareOperator.EQUAL, new BinaryComparator(ROWS[3]))); 313 int numberOfMatchingRowFilters = filters.size(); 314 filter = new FilterList(Operator.MUST_PASS_ONE, filters); 315 testRowsFilteredMetric(baseScan, filter, ROWS.length - numberOfMatchingRowFilters); 316 filters.clear(); 317 318 // Add a single column value exclude filter for each column... The net effect is that all 319 // columns will be excluded when scanning on the server side. This will result in an empty cell 320 // array in RegionScanner#nextInternal which should be interpreted as a row being filtered. 321 for (int family = 0; family < FAMILIES.length; family++) { 322 for (int qualifier = 0; qualifier < QUALIFIERS.length; qualifier++) { 323 filters.add(new SingleColumnValueExcludeFilter(FAMILIES[family], QUALIFIERS[qualifier], 324 CompareOperator.EQUAL, VALUE)); 325 } 326 } 327 filter = new FilterList(Operator.MUST_PASS_ONE, filters); 328 testRowsFilteredMetric(baseScan, filter, ROWS.length); 329 } 330 331 private void testRowsFilteredMetric(Scan baseScan, Filter filter, int expectedNumFiltered) 332 throws Exception { 333 Scan scan = new Scan(baseScan); 334 if (filter != null) { 335 scan.setFilter(filter); 336 } 337 testMetric(scan, ServerSideScanMetrics.COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME, 338 expectedNumFiltered); 339 } 340 341 /** 342 * Run the scan to completetion and check the metric against the specified value 343 * @param scan The scan instance to use to record metrics 344 * @param metricKey The metric key name 345 * @param expectedValue The expected value of metric 346 * @throws Exception on unexpected failure 347 */ 348 private void testMetric(Scan scan, String metricKey, long expectedValue) throws Exception { 349 testMetric(scan, metricKey, expectedValue, CompareOperator.EQUAL); 350 } 351 352 private void testMetric(Scan scan, String metricKey, long expectedValue, 353 CompareOperator compareOperator) throws Exception { 354 assertTrue("Scan should be configured to record metrics", scan.isScanMetricsEnabled()); 355 ResultScanner scanner = TABLE.getScanner(scan); 356 // Iterate through all the results 357 while (scanner.next() != null) { 358 continue; 359 } 360 scanner.close(); 361 ScanMetrics metrics = scanner.getScanMetrics(); 362 assertNotNull("Metrics are null", metrics); 363 assertTrue("Metric : " + metricKey + " does not exist", metrics.hasCounter(metricKey)); 364 final long actualMetricValue = metrics.getCounter(metricKey).get(); 365 if (compareOperator == CompareOperator.EQUAL) { 366 assertEquals( 367 "Metric: " + metricKey + " Expected: " + expectedValue + " Actual: " + actualMetricValue, 368 expectedValue, actualMetricValue); 369 } else { 370 assertTrue( 371 "Metric: " + metricKey + " Expected: > " + expectedValue + " Actual: " + actualMetricValue, 372 actualMetricValue > expectedValue); 373 } 374 } 375}