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 static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertTrue;
022
023import java.io.IOException;
024import java.util.ArrayList;
025import java.util.Arrays;
026import java.util.Collections;
027import java.util.HashMap;
028import java.util.HashSet;
029import java.util.List;
030import java.util.Map;
031import java.util.Random;
032import java.util.Set;
033import java.util.stream.Stream;
034import org.apache.hadoop.hbase.Cell;
035import org.apache.hadoop.hbase.CellComparatorImpl;
036import org.apache.hadoop.hbase.CellUtil;
037import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate;
038import org.apache.hadoop.hbase.HBaseTestingUtil;
039import org.apache.hadoop.hbase.HConstants;
040import org.apache.hadoop.hbase.KeyValue;
041import org.apache.hadoop.hbase.PrivateCellUtil;
042import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
043import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
044import org.apache.hadoop.hbase.client.Delete;
045import org.apache.hadoop.hbase.client.Put;
046import org.apache.hadoop.hbase.client.Scan;
047import org.apache.hadoop.hbase.filter.PrefixFilter;
048import org.apache.hadoop.hbase.io.compress.Compression;
049import org.apache.hadoop.hbase.testclassification.MediumTests;
050import org.apache.hadoop.hbase.testclassification.RegionServerTests;
051import org.apache.hadoop.hbase.util.BloomFilterUtil;
052import org.apache.hadoop.hbase.util.Bytes;
053import org.junit.jupiter.api.AfterEach;
054import org.junit.jupiter.api.BeforeEach;
055import org.junit.jupiter.api.Tag;
056import org.junit.jupiter.api.TestInfo;
057import org.junit.jupiter.api.TestTemplate;
058import org.junit.jupiter.params.provider.Arguments;
059import org.slf4j.Logger;
060import org.slf4j.LoggerFactory;
061
062/**
063 * Test various seek optimizations for correctness and check if they are actually saving I/O
064 * operations.
065 */
066@Tag(RegionServerTests.TAG)
067@Tag(MediumTests.TAG)
068@HBaseParameterizedTestTemplate(name = "{index}: comprAlgo={0}, bloomType={1}")
069public class TestSeekOptimizations {
070
071  private static final Logger LOG = LoggerFactory.getLogger(TestSeekOptimizations.class);
072
073  // Constants
074  private static final String FAMILY = "myCF";
075  private static final byte[] FAMILY_BYTES = Bytes.toBytes(FAMILY);
076
077  private static final int PUTS_PER_ROW_COL = 50;
078  private static final int DELETES_PER_ROW_COL = 10;
079
080  private static final int NUM_ROWS = 3;
081  private static final int NUM_COLS = 3;
082
083  private static final boolean VERBOSE = false;
084
085  /**
086   * Disable this when this test fails hopelessly and you need to debug a simpler case.
087   */
088  private static final boolean USE_MANY_STORE_FILES = true;
089
090  private static final int[][] COLUMN_SETS = new int[][] { {}, // All columns
091    { 0 }, { 1 }, { 0, 2 }, { 1, 2 }, { 0, 1, 2 }, };
092
093  // Both start row and end row are inclusive here for the purposes of this
094  // test.
095  private static final int[][] ROW_RANGES =
096    new int[][] { { -1, -1 }, { 0, 1 }, { 1, 1 }, { 1, 2 }, { 0, 2 } };
097
098  private static final int[] MAX_VERSIONS_VALUES = new int[] { 1, 2 };
099
100  // Instance variables
101  private HRegion region;
102  private Put put;
103  private Delete del;
104  private Set<Long> putTimestamps = new HashSet<>();
105  private Set<Long> delTimestamps = new HashSet<>();
106  private List<Cell> expectedKVs = new ArrayList<>();
107
108  private Compression.Algorithm comprAlgo;
109  private BloomType bloomType;
110
111  private long totalSeekDiligent, totalSeekLazy;
112
113  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
114  private static final Random RNG = new Random(); // This test depends on Random#setSeed
115
116  public static Stream<Arguments> parameters() {
117    return HBaseTestingUtil.BLOOM_AND_COMPRESSION_COMBINATIONS.stream().map(Arguments::of);
118  }
119
120  public TestSeekOptimizations(Compression.Algorithm comprAlgo, BloomType bloomType) {
121    this.comprAlgo = comprAlgo;
122    this.bloomType = bloomType;
123  }
124
125  @BeforeEach
126  public void setUp(TestInfo testInfo) throws IOException {
127    RNG.setSeed(91238123L);
128    expectedKVs.clear();
129    TEST_UTIL.getConfiguration().setInt(BloomFilterUtil.PREFIX_LENGTH_KEY, 10);
130
131    // enable seek counting
132    StoreFileScanner.instrument();
133    ColumnFamilyDescriptor columnFamilyDescriptor =
134      ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(FAMILY)).setCompressionType(comprAlgo)
135        .setBloomFilterType(bloomType).setMaxVersions(3).build();
136
137    region =
138      TEST_UTIL.createTestRegion(testInfo.getTestMethod().get().getName(), columnFamilyDescriptor);
139
140    // Delete the given timestamp and everything before.
141    final long latestDelTS = USE_MANY_STORE_FILES ? 1397 : -1;
142
143    createTimestampRange(1, 50, -1);
144    createTimestampRange(51, 100, -1);
145    if (USE_MANY_STORE_FILES) {
146      createTimestampRange(100, 500, 127);
147      createTimestampRange(900, 1300, -1);
148      createTimestampRange(1301, 2500, latestDelTS);
149      createTimestampRange(2502, 2598, -1);
150      createTimestampRange(2599, 2999, -1);
151    }
152
153    prepareExpectedKVs(latestDelTS);
154  }
155
156  @TestTemplate
157  public void testMultipleTimestampRanges() throws IOException {
158    for (int[] columnArr : COLUMN_SETS) {
159      for (int[] rowRange : ROW_RANGES) {
160        for (int maxVersions : MAX_VERSIONS_VALUES) {
161          for (boolean lazySeekEnabled : new boolean[] { false, true }) {
162            testScan(columnArr, lazySeekEnabled, rowRange[0], rowRange[1], maxVersions, false);
163          }
164        }
165      }
166    }
167
168    final double seekSavings = 1 - totalSeekLazy * 1.0 / totalSeekDiligent;
169    System.err.println("For bloom=" + bloomType + ", compr=" + comprAlgo
170      + " total seeks without optimization: " + totalSeekDiligent + ", with optimization: "
171      + totalSeekLazy + " (" + String.format("%.2f%%", totalSeekLazy * 100.0 / totalSeekDiligent)
172      + "), savings: " + String.format("%.2f%%", 100.0 * seekSavings) + "\n");
173
174    // Test that lazy seeks are buying us something. Without the actual
175    // implementation of the lazy seek optimization this will be 0.
176    final double expectedSeekSavings = 0.0;
177    assertTrue(seekSavings >= expectedSeekSavings,
178      "Lazy seek is only saving " + String.format("%.2f%%", seekSavings * 100)
179        + " seeks but should " + "save at least "
180        + String.format("%.2f%%", expectedSeekSavings * 100));
181  }
182
183  private ScanResult testScan(final int[] columnArr, final boolean lazySeekEnabled,
184    final int startRow, final int endRow, final int maxVersions, final boolean filtered)
185    throws IOException {
186    StoreScanner.enableLazySeekGlobally(lazySeekEnabled);
187    final Scan scan = new Scan();
188    final Set<String> qualSet = new HashSet<>();
189    for (int iColumn : columnArr) {
190      String qualStr = getQualStr(iColumn);
191      scan.addColumn(FAMILY_BYTES, Bytes.toBytes(qualStr));
192      qualSet.add(qualStr);
193    }
194    if (filtered) {
195      scan.setFilter(new PrefixFilter(Bytes.toBytes("row")));
196    }
197    scan.readVersions(maxVersions);
198    scan.withStartRow(rowBytes(startRow));
199
200    // Adjust for the fact that for multi-row queries the end row is exclusive.
201    if (startRow != endRow) {
202      scan.withStopRow(rowBytes(endRow + 1));
203    } else {
204      scan.withStopRow(rowBytes(endRow), true);
205    }
206
207    final long initialSeekCount = StoreFileScanner.getSeekCount();
208    final InternalScanner scanner = region.getScanner(scan);
209    final long scannerOpenSeekCount = StoreFileScanner.getSeekCount() - initialSeekCount;
210    final List<Cell> results = new ArrayList<>();
211    final List<Cell> actualKVs = new ArrayList<>();
212
213    // Such a clumsy do-while loop appears to be the official way to use an
214    // internalScanner. scanner.next() return value refers to the _next_
215    // result, not to the one already returned in results.
216    try {
217      boolean hasNext;
218      do {
219        hasNext = scanner.next(results);
220        actualKVs.addAll(results);
221        results.clear();
222      } while (hasNext);
223    } finally {
224      scanner.close();
225    }
226
227    List<Cell> filteredKVs =
228      filterExpectedResults(qualSet, rowBytes(startRow), rowBytes(endRow), maxVersions);
229    final String rowRestrictionStr = (startRow == -1 && endRow == -1)
230      ? "all rows"
231      : (startRow == endRow
232        ? ("row=" + startRow)
233        : ("startRow=" + startRow + ", " + "endRow=" + endRow));
234    final String columnRestrictionStr =
235      columnArr.length == 0 ? "all columns" : ("columns=" + Arrays.toString(columnArr));
236    final String testDesc = "Bloom=" + bloomType + ", compr=" + comprAlgo + ", "
237      + (scan.isGetScan() ? "Get" : "Scan") + ": " + columnRestrictionStr + ", " + rowRestrictionStr
238      + ", maxVersions=" + maxVersions + ", lazySeek=" + lazySeekEnabled;
239    long seekCount = StoreFileScanner.getSeekCount() - initialSeekCount;
240    if (VERBOSE) {
241      System.err.println("Seek count: " + seekCount + ", KVs returned: " + actualKVs.size() + ". "
242        + testDesc + (lazySeekEnabled ? "\n" : ""));
243    }
244    if (lazySeekEnabled) {
245      totalSeekLazy += seekCount;
246    } else {
247      totalSeekDiligent += seekCount;
248    }
249    assertKVListsEqual(testDesc, filteredKVs, actualKVs);
250    return new ScanResult(actualKVs, scannerOpenSeekCount);
251  }
252
253  private List<Cell> filterExpectedResults(Set<String> qualSet, byte[] startRow, byte[] endRow,
254    int maxVersions) {
255    final List<Cell> filteredKVs = new ArrayList<>();
256    final Map<String, Integer> verCount = new HashMap<>();
257    for (Cell kv : expectedKVs) {
258      if (
259        startRow.length > 0 && Bytes.compareTo(kv.getRowArray(), kv.getRowOffset(),
260          kv.getRowLength(), startRow, 0, startRow.length) < 0
261      ) {
262        continue;
263      }
264
265      // In this unit test the end row is always inclusive.
266      if (
267        endRow.length > 0 && Bytes.compareTo(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength(),
268          endRow, 0, endRow.length) > 0
269      ) {
270        continue;
271      }
272
273      if (
274        !qualSet.isEmpty() && (!CellUtil.matchingFamily(kv, FAMILY_BYTES)
275          || !qualSet.contains(Bytes.toString(CellUtil.cloneQualifier(kv))))
276      ) {
277        continue;
278      }
279
280      final String rowColStr = Bytes.toStringBinary(CellUtil.cloneRow(kv)) + "/"
281        + Bytes.toStringBinary(CellUtil.cloneFamily(kv)) + ":"
282        + Bytes.toStringBinary(CellUtil.cloneQualifier(kv));
283      final Integer curNumVer = verCount.get(rowColStr);
284      final int newNumVer = curNumVer != null ? (curNumVer + 1) : 1;
285      if (newNumVer <= maxVersions) {
286        filteredKVs.add(kv);
287        verCount.put(rowColStr, newNumVer);
288      }
289    }
290
291    return filteredKVs;
292  }
293
294  private void prepareExpectedKVs(long latestDelTS) {
295    final List<Cell> filteredKVs = new ArrayList<>();
296    for (Cell kv : expectedKVs) {
297      if (kv.getTimestamp() > latestDelTS || latestDelTS == -1) {
298        filteredKVs.add(kv);
299      }
300    }
301    expectedKVs = filteredKVs;
302    Collections.sort(expectedKVs, CellComparatorImpl.COMPARATOR);
303  }
304
305  public void put(String qual, long ts) {
306    if (!putTimestamps.contains(ts)) {
307      put.addColumn(FAMILY_BYTES, Bytes.toBytes(qual), ts, createValue(ts));
308      putTimestamps.add(ts);
309    }
310    if (VERBOSE) {
311      LOG.info("put: row " + Bytes.toStringBinary(put.getRow()) + ", cf " + FAMILY + ", qualifier "
312        + qual + ", ts " + ts);
313    }
314  }
315
316  private byte[] createValue(long ts) {
317    return Bytes.toBytes("value" + ts);
318  }
319
320  public void delAtTimestamp(String qual, long ts) {
321    del.addColumn(FAMILY_BYTES, Bytes.toBytes(qual), ts);
322    logDelete(qual, ts, "at");
323  }
324
325  private void logDelete(String qual, long ts, String delType) {
326    if (VERBOSE) {
327      LOG.info("del " + delType + ": row " + Bytes.toStringBinary(put.getRow()) + ", cf " + FAMILY
328        + ", qualifier " + qual + ", ts " + ts);
329    }
330  }
331
332  private void delUpToTimestamp(String qual, long upToTS) {
333    del.addColumns(FAMILY_BYTES, Bytes.toBytes(qual), upToTS);
334    logDelete(qual, upToTS, "up to and including");
335  }
336
337  private long randLong(long n) {
338    long l = RNG.nextLong();
339    if (l == Long.MIN_VALUE) l = Long.MAX_VALUE;
340    return Math.abs(l) % n;
341  }
342
343  private long randBetween(long a, long b) {
344    long x = a + randLong(b - a + 1);
345    assertTrue(a <= x && x <= b);
346    return x;
347  }
348
349  private final String rowStr(int i) {
350    return ("row" + i).intern();
351  }
352
353  private final byte[] rowBytes(int i) {
354    if (i == -1) {
355      return HConstants.EMPTY_BYTE_ARRAY;
356    }
357    return Bytes.toBytes(rowStr(i));
358  }
359
360  private final String getQualStr(int i) {
361    return ("qual" + i).intern();
362  }
363
364  public void createTimestampRange(long minTS, long maxTS, long deleteUpToTS) throws IOException {
365    assertTrue(minTS < maxTS);
366    assertTrue(deleteUpToTS == -1 || (minTS <= deleteUpToTS && deleteUpToTS <= maxTS));
367
368    for (int iRow = 0; iRow < NUM_ROWS; ++iRow) {
369      final String row = rowStr(iRow);
370      final byte[] rowBytes = Bytes.toBytes(row);
371      for (int iCol = 0; iCol < NUM_COLS; ++iCol) {
372        final String qual = getQualStr(iCol);
373        final byte[] qualBytes = Bytes.toBytes(qual);
374        put = new Put(rowBytes);
375
376        putTimestamps.clear();
377        put(qual, minTS);
378        put(qual, maxTS);
379        for (int i = 0; i < PUTS_PER_ROW_COL; ++i) {
380          put(qual, randBetween(minTS, maxTS));
381        }
382
383        long[] putTimestampList = new long[putTimestamps.size()];
384        {
385          int i = 0;
386          for (long ts : putTimestamps) {
387            putTimestampList[i++] = ts;
388          }
389        }
390
391        // Delete a predetermined number of particular timestamps
392        delTimestamps.clear();
393        assertTrue(putTimestampList.length >= DELETES_PER_ROW_COL);
394        int numToDel = DELETES_PER_ROW_COL;
395        int tsRemaining = putTimestampList.length;
396        del = new Delete(rowBytes);
397        for (long ts : putTimestampList) {
398          if (RNG.nextInt(tsRemaining) < numToDel) {
399            delAtTimestamp(qual, ts);
400            putTimestamps.remove(ts);
401            --numToDel;
402          }
403
404          if (--tsRemaining == 0) {
405            break;
406          }
407        }
408
409        // Another type of delete: everything up to the given timestamp.
410        if (deleteUpToTS != -1) {
411          delUpToTimestamp(qual, deleteUpToTS);
412        }
413
414        region.put(put);
415        if (!del.isEmpty()) {
416          region.delete(del);
417        }
418
419        // Add remaining timestamps (those we have not deleted) to expected
420        // results
421        for (long ts : putTimestamps) {
422          expectedKVs.add(new KeyValue(rowBytes, FAMILY_BYTES, qualBytes, ts, KeyValue.Type.Put));
423        }
424      }
425    }
426
427    region.flush(true);
428  }
429
430  @AfterEach
431  public void tearDown() throws IOException {
432    if (region != null) {
433      HBaseTestingUtil.closeRegionAndWAL(region);
434    }
435
436    // We have to re-set the lazy seek flag back to the default so that other
437    // unit tests are not affected.
438    StoreScanner.enableLazySeekGlobally(StoreScanner.LAZY_SEEK_ENABLED_BY_DEFAULT);
439  }
440
441  public void assertKVListsEqual(String additionalMsg, final List<? extends Cell> expected,
442    final List<? extends Cell> actual) {
443    final int eLen = expected.size();
444    final int aLen = actual.size();
445    final int minLen = Math.min(eLen, aLen);
446
447    int i;
448    for (i = 0; i < minLen && PrivateCellUtil.compareKeyIgnoresMvcc(CellComparatorImpl.COMPARATOR,
449      expected.get(i), actual.get(i)) == 0; ++i) {
450    }
451
452    if (additionalMsg == null) {
453      additionalMsg = "";
454    }
455    if (!additionalMsg.isEmpty()) {
456      additionalMsg = ". " + additionalMsg;
457    }
458
459    if (eLen != aLen || i != minLen) {
460      throw new AssertionError("Expected and actual KV arrays differ at position " + i + ": "
461        + HBaseTestingUtil.safeGetAsStr(expected, i) + " (length " + eLen + ") vs. "
462        + HBaseTestingUtil.safeGetAsStr(actual, i) + " (length " + aLen + ")" + additionalMsg);
463    }
464  }
465
466  @TestTemplate
467  public void testSeeksEagerlyWhenFiltered() throws IOException {
468    ScanResult filteredLazyResults = testScan(new int[] { 0 }, true, 0, 2, 1, true);
469    ScanResult filteredEagerResults = testScan(new int[] { 0 }, false, 0, 2, 1, true);
470    assertKVListsEqual("Filtered explicit column scan results differ with lazy seeking enabled",
471      filteredEagerResults.cells, filteredLazyResults.cells);
472    assertEquals(filteredEagerResults.scannerOpenSeekCount,
473      filteredLazyResults.scannerOpenSeekCount,
474      "Filtered explicit column scans must always eagerly seek");
475  }
476
477  private static final class ScanResult {
478    private final List<Cell> cells;
479    private final long scannerOpenSeekCount;
480
481    private ScanResult(List<Cell> cells, long scannerOpenSeekCount) {
482      this.cells = cells;
483      this.scannerOpenSeekCount = scannerOpenSeekCount;
484    }
485  }
486}