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}