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