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.apache.hadoop.hbase.HBaseTestingUtil.START_KEY_BYTES;
021import static org.apache.hadoop.hbase.HBaseTestingUtil.fam1;
022import static org.apache.hadoop.hbase.HBaseTestingUtil.fam2;
023import static org.junit.jupiter.api.Assertions.assertEquals;
024import static org.junit.jupiter.api.Assertions.assertFalse;
025import static org.junit.jupiter.api.Assertions.assertNotNull;
026import static org.junit.jupiter.api.Assertions.assertTrue;
027import static org.junit.jupiter.api.Assertions.fail;
028
029import java.io.IOException;
030import java.util.ArrayList;
031import java.util.List;
032import java.util.NavigableSet;
033import org.apache.hadoop.hbase.Cell;
034import org.apache.hadoop.hbase.CellUtil;
035import org.apache.hadoop.hbase.CompareOperator;
036import org.apache.hadoop.hbase.ExtendedCell;
037import org.apache.hadoop.hbase.HBaseTestingUtil;
038import org.apache.hadoop.hbase.HConstants;
039import org.apache.hadoop.hbase.HTestConst;
040import org.apache.hadoop.hbase.TableName;
041import org.apache.hadoop.hbase.UnknownScannerException;
042import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
043import org.apache.hadoop.hbase.client.Delete;
044import org.apache.hadoop.hbase.client.Get;
045import org.apache.hadoop.hbase.client.Put;
046import org.apache.hadoop.hbase.client.RegionInfo;
047import org.apache.hadoop.hbase.client.RegionInfoBuilder;
048import org.apache.hadoop.hbase.client.Result;
049import org.apache.hadoop.hbase.client.ResultScanner;
050import org.apache.hadoop.hbase.client.Scan;
051import org.apache.hadoop.hbase.client.Table;
052import org.apache.hadoop.hbase.client.TableDescriptor;
053import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
054import org.apache.hadoop.hbase.filter.BinaryComparator;
055import org.apache.hadoop.hbase.filter.ByteArrayComparable;
056import org.apache.hadoop.hbase.filter.Filter;
057import org.apache.hadoop.hbase.filter.InclusiveStopFilter;
058import org.apache.hadoop.hbase.filter.PrefixFilter;
059import org.apache.hadoop.hbase.filter.RowFilter;
060import org.apache.hadoop.hbase.filter.WhileMatchFilter;
061import org.apache.hadoop.hbase.testclassification.MediumTests;
062import org.apache.hadoop.hbase.testclassification.RegionServerTests;
063import org.apache.hadoop.hbase.util.Bytes;
064import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
065import org.junit.jupiter.api.Tag;
066import org.junit.jupiter.api.Test;
067import org.junit.jupiter.api.TestInfo;
068import org.slf4j.Logger;
069import org.slf4j.LoggerFactory;
070
071/**
072 * Test of a long-lived scanner validating as we go.
073 */
074@Tag(RegionServerTests.TAG)
075@Tag(MediumTests.TAG)
076public class TestScanner {
077
078  private static final Logger LOG = LoggerFactory.getLogger(TestScanner.class);
079  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
080
081  private static final byte[] FIRST_ROW = HConstants.EMPTY_START_ROW;
082  private static final byte[][] COLS = { HConstants.CATALOG_FAMILY };
083  private static final byte[][] EXPLICIT_COLS =
084    { HConstants.REGIONINFO_QUALIFIER, HConstants.SERVER_QUALIFIER,
085    // TODO ryan
086    // HConstants.STARTCODE_QUALIFIER
087    };
088
089  static final TableDescriptor TESTTABLEDESC =
090    TableDescriptorBuilder.newBuilder(TableName.valueOf("testscanner"))
091      .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(HConstants.CATALOG_FAMILY)
092        // Ten is an arbitrary number. Keep versions to help debugging.
093        .setMaxVersions(10).setBlockCacheEnabled(false).setBlocksize(8 * 1024).build())
094      .build();
095
096  /** HRegionInfo for root region */
097  public static final RegionInfo REGION_INFO =
098    RegionInfoBuilder.newBuilder(TESTTABLEDESC.getTableName()).build();
099
100  private static final byte[] ROW_KEY = REGION_INFO.getRegionName();
101
102  private static final long START_CODE = Long.MAX_VALUE;
103
104  private static final byte[] LAZY_SEEK_FAMILY = Bytes.toBytes("family");
105  private static final byte[] LAZY_SEEK_QUALIFIER = Bytes.toBytes("qualifier");
106  private static final byte[] LAZY_SEEK_ROW = Bytes.toBytes("row");
107
108  private HRegion region;
109
110  private byte[] firstRowBytes, secondRowBytes, thirdRowBytes;
111  final private byte[] col1;
112
113  public TestScanner() {
114    super();
115
116    firstRowBytes = START_KEY_BYTES;
117    secondRowBytes = START_KEY_BYTES.clone();
118    // Increment the least significant character so we get to next row.
119    secondRowBytes[START_KEY_BYTES.length - 1]++;
120    thirdRowBytes = START_KEY_BYTES.clone();
121    thirdRowBytes[START_KEY_BYTES.length - 1] =
122      (byte) (thirdRowBytes[START_KEY_BYTES.length - 1] + 2);
123    col1 = Bytes.toBytes("column1");
124  }
125
126  private static final class RecordingStoreScanner extends StoreScanner {
127    private boolean initialSeekWasLazy;
128
129    RecordingStoreScanner(HStore store, Scan scan, NavigableSet<byte[]> columns)
130      throws IOException {
131      super(store, store.getScanInfo(), scan, columns, Long.MAX_VALUE);
132    }
133
134    @Override
135    protected void seekScanners(List<? extends KeyValueScanner> scanners, ExtendedCell seekKey,
136      boolean isLazy, boolean isParallelSeek) throws IOException {
137      initialSeekWasLazy = isLazy;
138      super.seekScanners(scanners, seekKey, isLazy, isParallelSeek);
139    }
140  }
141
142  private static final class TrackingRowComparator extends ByteArrayComparable {
143    private final List<byte[]> comparedRows = new ArrayList<>();
144
145    TrackingRowComparator(byte[] value) {
146      super(value);
147    }
148
149    @Override
150    public int compareTo(byte[] value, int offset, int length) {
151      comparedRows.add(Bytes.copy(value, offset, length));
152      return Bytes.compareTo(getValue(), 0, getValue().length, value, offset, length);
153    }
154
155    @Override
156    public byte[] toByteArray() {
157      return getValue();
158    }
159  }
160
161  @Test
162  public void testFilterComparatorOnlySeesActualRows() throws Exception {
163    byte[] family = Bytes.toBytes("family");
164    byte[] qualifier = Bytes.toBytes("qualifier");
165    byte[] regionStartKey = new byte[] { 1 };
166    byte[] row = new byte[] { 1, 0, 1 };
167    TableDescriptor tableDescriptor =
168      TableDescriptorBuilder.newBuilder(TableName.valueOf("testFilterComparatorOnlySeesActualRows"))
169        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(family)
170          .setBloomFilterType(BloomType.ROWCOL).build())
171        .build();
172    TrackingRowComparator comparator = new TrackingRowComparator(row);
173
174    StoreScanner.enableLazySeekGlobally(true);
175    try {
176      this.region = TEST_UTIL.createLocalHRegion(tableDescriptor, regionStartKey, null);
177      Put put = new Put(row);
178      put.addColumn(family, qualifier, Bytes.toBytes("value"));
179      region.put(put);
180      region.flush(true);
181
182      Scan scan = new Scan().withStartRow(regionStartKey);
183      scan.addColumn(family, qualifier);
184      scan.setFilter(new RowFilter(CompareOperator.EQUAL, comparator));
185      List<Cell> results = new ArrayList<>();
186      try (InternalScanner scanner = region.getScanner(scan)) {
187        assertFalse(scanner.next(results));
188      }
189
190      assertEquals(1, results.size());
191      assertTrue(CellUtil.matchingRows(results.get(0), row));
192      assertEquals(1, comparator.comparedRows.size());
193      assertTrue(Bytes.equals(row, comparator.comparedRows.get(0)));
194    } finally {
195      StoreScanner.enableLazySeekGlobally(StoreScanner.LAZY_SEEK_ENABLED_BY_DEFAULT);
196      HBaseTestingUtil.closeRegionAndWAL(this.region);
197    }
198  }
199
200  @Test
201  public void testWhileMatchFilterOnlySeesActualRows() throws Exception {
202    byte[] family = Bytes.toBytes("family");
203    byte[] qualifier = Bytes.toBytes("qualifier");
204    TableDescriptor tableDescriptor =
205      TableDescriptorBuilder.newBuilder(TableName.valueOf("testWhileMatchFilterOnlySeesActualRows"))
206        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(family).build()).build();
207
208    StoreScanner.enableLazySeekGlobally(true);
209    try {
210      this.region = TEST_UTIL.createLocalHRegion(tableDescriptor, null, null);
211      List<String> rows = List.of("row1", "row2", "row3");
212      for (String row : rows) {
213        Put put = new Put(Bytes.toBytes(row)).addColumn(family, qualifier, Bytes.toBytes("value"));
214        region.put(put);
215      }
216      region.flush(true);
217
218      Scan scan = new Scan().addColumn(family, qualifier)
219        .setFilter(new WhileMatchFilter(new RowFilter(CompareOperator.NOT_EQUAL,
220          new BinaryComparator(HConstants.EMPTY_START_ROW))));
221      int scannedRows = 0;
222      try (InternalScanner scanner = region.getScanner(scan)) {
223        boolean hasMoreRows;
224        do {
225          List<Cell> results = new ArrayList<>();
226          hasMoreRows = scanner.next(results);
227          if (!results.isEmpty()) {
228            ++scannedRows;
229          }
230        } while (hasMoreRows);
231      }
232
233      assertEquals(rows.size(), scannedRows);
234    } finally {
235      StoreScanner.enableLazySeekGlobally(StoreScanner.LAZY_SEEK_ENABLED_BY_DEFAULT);
236      HBaseTestingUtil.closeRegionAndWAL(this.region);
237    }
238  }
239
240  @Test
241  public void testInitialLazySeekForUnfilteredExplicitColumnScan() throws Exception {
242    Scan scan = new Scan().withStartRow(LAZY_SEEK_ROW);
243    scan.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER);
244    assertInitialLazySeek(scan, true, true);
245  }
246
247  @Test
248  public void testInitialLazySeekForFilteredGet() throws Exception {
249    Get get = new Get(LAZY_SEEK_ROW);
250    get.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER);
251    get.setFilter(new PrefixFilter(LAZY_SEEK_ROW));
252    assertInitialLazySeek(new Scan(get), true, true);
253  }
254
255  @Test
256  public void testInitialLazySeekForFilteredNonGetScan() throws Exception {
257    Scan scan = new Scan().withStartRow(LAZY_SEEK_ROW);
258    scan.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER);
259    scan.setFilter(new PrefixFilter(LAZY_SEEK_ROW));
260    assertInitialLazySeek(scan, true, false);
261  }
262
263  @Test
264  public void testInitialLazySeekForAllColumnScan() throws Exception {
265    assertInitialLazySeek(new Scan().withStartRow(LAZY_SEEK_ROW), true, false);
266  }
267
268  @Test
269  public void testInitialLazySeekWhenDisabledGlobally() throws Exception {
270    Scan scan = new Scan().withStartRow(LAZY_SEEK_ROW);
271    scan.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER);
272    assertInitialLazySeek(scan, false, false);
273  }
274
275  private void assertInitialLazySeek(Scan scan, boolean lazySeekEnabled, boolean expected)
276    throws IOException {
277    StoreScanner.enableLazySeekGlobally(lazySeekEnabled);
278    try {
279      HStore store = createLazySeekTestStore();
280      try (RecordingStoreScanner scanner =
281        new RecordingStoreScanner(store, scan, scan.getFamilyMap().get(LAZY_SEEK_FAMILY))) {
282        assertEquals(expected, scanner.initialSeekWasLazy);
283      }
284    } finally {
285      StoreScanner.enableLazySeekGlobally(StoreScanner.LAZY_SEEK_ENABLED_BY_DEFAULT);
286      if (this.region != null) {
287        HBaseTestingUtil.closeRegionAndWAL(this.region);
288        this.region = null;
289      }
290    }
291  }
292
293  private HStore createLazySeekTestStore() throws IOException {
294    TableDescriptor tableDescriptor = TableDescriptorBuilder
295      .newBuilder(TableName.valueOf("testInitialLazySeek"))
296      .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(LAZY_SEEK_FAMILY).build()).build();
297    this.region = TEST_UTIL.createLocalHRegion(tableDescriptor, null, null);
298    Put put = new Put(LAZY_SEEK_ROW);
299    put.addColumn(LAZY_SEEK_FAMILY, LAZY_SEEK_QUALIFIER, Bytes.toBytes("value"));
300    region.put(put);
301    region.flush(true);
302    return region.getStore(LAZY_SEEK_FAMILY);
303  }
304
305  /**
306   * Test basic stop row filter works.
307   */
308  @Test
309  public void testStopRow() throws Exception {
310    byte[] startrow = Bytes.toBytes("bbb");
311    byte[] stoprow = Bytes.toBytes("ccc");
312    try {
313      this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null);
314      HTestConst.addContent(this.region, HConstants.CATALOG_FAMILY);
315      List<Cell> results = new ArrayList<>();
316      // Do simple test of getting one row only first.
317      Scan scan = new Scan().withStartRow(Bytes.toBytes("abc")).withStopRow(Bytes.toBytes("abd"));
318      scan.addFamily(HConstants.CATALOG_FAMILY);
319
320      InternalScanner s = region.getScanner(scan);
321      int count = 0;
322      while (s.next(results)) {
323        count++;
324      }
325      s.close();
326      assertEquals(0, count);
327      // Now do something a bit more imvolved.
328      scan = new Scan().withStartRow(startrow).withStopRow(stoprow);
329      scan.addFamily(HConstants.CATALOG_FAMILY);
330
331      s = region.getScanner(scan);
332      count = 0;
333      Cell kv = null;
334      results = new ArrayList<>();
335      for (boolean first = true; s.next(results);) {
336        kv = results.get(0);
337        if (first) {
338          assertTrue(CellUtil.matchingRows(kv, startrow));
339          first = false;
340        }
341        count++;
342      }
343      assertTrue(Bytes.BYTES_COMPARATOR.compare(stoprow, CellUtil.cloneRow(kv)) > 0);
344      // We got something back.
345      assertTrue(count > 10);
346      s.close();
347    } finally {
348      HBaseTestingUtil.closeRegionAndWAL(this.region);
349    }
350  }
351
352  void rowPrefixFilter(Scan scan) throws IOException {
353    List<Cell> results = new ArrayList<>();
354    scan.addFamily(HConstants.CATALOG_FAMILY);
355    InternalScanner s = region.getScanner(scan);
356    boolean hasMore = true;
357    while (hasMore) {
358      hasMore = s.next(results);
359      for (Cell kv : results) {
360        assertEquals((byte) 'a', CellUtil.cloneRow(kv)[0]);
361        assertEquals((byte) 'b', CellUtil.cloneRow(kv)[1]);
362      }
363      results.clear();
364    }
365    s.close();
366  }
367
368  void rowInclusiveStopFilter(Scan scan, byte[] stopRow) throws IOException {
369    List<Cell> results = new ArrayList<>();
370    scan.addFamily(HConstants.CATALOG_FAMILY);
371    InternalScanner s = region.getScanner(scan);
372    boolean hasMore = true;
373    while (hasMore) {
374      hasMore = s.next(results);
375      for (Cell kv : results) {
376        assertTrue(Bytes.compareTo(CellUtil.cloneRow(kv), stopRow) <= 0);
377      }
378      results.clear();
379    }
380    s.close();
381  }
382
383  @Test
384  public void testFilters() throws IOException {
385    try {
386      this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null);
387      HTestConst.addContent(this.region, HConstants.CATALOG_FAMILY);
388      byte[] prefix = Bytes.toBytes("ab");
389      Filter newFilter = new PrefixFilter(prefix);
390      Scan scan = new Scan();
391      scan.setFilter(newFilter);
392      rowPrefixFilter(scan);
393
394      byte[] stopRow = Bytes.toBytes("bbc");
395      newFilter = new WhileMatchFilter(new InclusiveStopFilter(stopRow));
396      scan = new Scan();
397      scan.setFilter(newFilter);
398      rowInclusiveStopFilter(scan, stopRow);
399
400    } finally {
401      HBaseTestingUtil.closeRegionAndWAL(this.region);
402    }
403  }
404
405  /**
406   * Test that closing a scanner while a client is using it doesn't throw NPEs but instead a
407   * UnknownScannerException. HBASE-2503
408   */
409  @Test
410  public void testRaceBetweenClientAndTimeout() throws Exception {
411    try {
412      this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null);
413      HTestConst.addContent(this.region, HConstants.CATALOG_FAMILY);
414      Scan scan = new Scan();
415      InternalScanner s = region.getScanner(scan);
416      List<Cell> results = new ArrayList<>();
417      try {
418        s.next(results);
419        s.close();
420        s.next(results);
421        fail("We don't want anything more, we should be failing");
422      } catch (UnknownScannerException ex) {
423        // ok!
424        return;
425      }
426    } finally {
427      HBaseTestingUtil.closeRegionAndWAL(this.region);
428    }
429  }
430
431  /**
432   * The test!
433   */
434  @Test
435  public void testScanner() throws IOException {
436    try {
437      region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null);
438      Table table = new RegionAsTable(region);
439
440      // Write information to the meta table
441
442      Put put = new Put(ROW_KEY, EnvironmentEdgeManager.currentTime());
443
444      put.addColumn(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER,
445        RegionInfo.toByteArray(REGION_INFO));
446      table.put(put);
447
448      // What we just committed is in the memstore. Verify that we can get
449      // it back both with scanning and get
450
451      scan(false, null);
452      getRegionInfo(table);
453
454      // Close and re-open
455
456      ((HRegion) region).close();
457      region = HRegion.openHRegion(region, null);
458      table = new RegionAsTable(region);
459
460      // Verify we can get the data back now that it is on disk.
461
462      scan(false, null);
463      getRegionInfo(table);
464
465      // Store some new information
466
467      String address = HConstants.LOCALHOST_IP + ":" + HBaseTestingUtil.randomFreePort();
468
469      put = new Put(ROW_KEY, EnvironmentEdgeManager.currentTime());
470      put.addColumn(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER, Bytes.toBytes(address));
471
472      // put.add(HConstants.COL_STARTCODE, Bytes.toBytes(START_CODE));
473
474      table.put(put);
475
476      // Validate that we can still get the HRegionInfo, even though it is in
477      // an older row on disk and there is a newer row in the memstore
478
479      scan(true, address.toString());
480      getRegionInfo(table);
481
482      // flush cache
483      this.region.flush(true);
484
485      // Validate again
486
487      scan(true, address.toString());
488      getRegionInfo(table);
489
490      // Close and reopen
491
492      ((HRegion) region).close();
493      region = HRegion.openHRegion(region, null);
494      table = new RegionAsTable(region);
495
496      // Validate again
497
498      scan(true, address.toString());
499      getRegionInfo(table);
500
501      // Now update the information again
502
503      address = "bar.foo.com:4321";
504
505      put = new Put(ROW_KEY, EnvironmentEdgeManager.currentTime());
506
507      put.addColumn(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER, Bytes.toBytes(address));
508      table.put(put);
509
510      // Validate again
511
512      scan(true, address.toString());
513      getRegionInfo(table);
514
515      // flush cache
516
517      region.flush(true);
518
519      // Validate again
520
521      scan(true, address.toString());
522      getRegionInfo(table);
523
524      // Close and reopen
525
526      ((HRegion) this.region).close();
527      this.region = HRegion.openHRegion(region, null);
528      table = new RegionAsTable(this.region);
529
530      // Validate again
531
532      scan(true, address.toString());
533      getRegionInfo(table);
534
535    } finally {
536      // clean up
537      HBaseTestingUtil.closeRegionAndWAL(this.region);
538    }
539  }
540
541  /** Compare the HRegionInfo we read from HBase to what we stored */
542  private void validateRegionInfo(byte[] regionBytes) throws IOException {
543    RegionInfo info = RegionInfo.parseFromOrNull(regionBytes);
544
545    assertEquals(REGION_INFO.getRegionId(), info.getRegionId());
546    assertEquals(0, info.getStartKey().length);
547    assertEquals(0, info.getEndKey().length);
548    assertEquals(0, Bytes.compareTo(info.getRegionName(), REGION_INFO.getRegionName()));
549    // assertEquals(0, info.getTableDesc().compareTo(REGION_INFO.getTableDesc()));
550  }
551
552  /** Use a scanner to get the region info and then validate the results */
553  private void scan(boolean validateStartcode, String serverName) throws IOException {
554    InternalScanner scanner = null;
555    Scan scan = null;
556    List<Cell> results = new ArrayList<>();
557    byte[][][] scanColumns = { COLS, EXPLICIT_COLS };
558    for (int i = 0; i < scanColumns.length; i++) {
559      try {
560        scan = new Scan().withStartRow(FIRST_ROW);
561        for (int ii = 0; ii < EXPLICIT_COLS.length; ii++) {
562          scan.addColumn(COLS[0], EXPLICIT_COLS[ii]);
563        }
564        scanner = region.getScanner(scan);
565        while (scanner.next(results)) {
566          assertTrue(
567            hasColumn(results, HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER));
568          byte[] val = CellUtil.cloneValue(
569            getColumn(results, HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER));
570          validateRegionInfo(val);
571          if (validateStartcode) {
572            // assertTrue(hasColumn(results, HConstants.CATALOG_FAMILY,
573            // HConstants.STARTCODE_QUALIFIER));
574            // val = getColumn(results, HConstants.CATALOG_FAMILY,
575            // HConstants.STARTCODE_QUALIFIER).getValue();
576            assertNotNull(val);
577            assertFalse(val.length == 0);
578            long startCode = Bytes.toLong(val);
579            assertEquals(START_CODE, startCode);
580          }
581
582          if (serverName != null) {
583            assertTrue(hasColumn(results, HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER));
584            val = CellUtil.cloneValue(
585              getColumn(results, HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER));
586            assertNotNull(val);
587            assertFalse(val.length == 0);
588            String server = Bytes.toString(val);
589            assertEquals(0, server.compareTo(serverName));
590          }
591        }
592      } finally {
593        InternalScanner s = scanner;
594        scanner = null;
595        if (s != null) {
596          s.close();
597        }
598      }
599    }
600  }
601
602  private boolean hasColumn(final List<Cell> kvs, final byte[] family, final byte[] qualifier) {
603    for (Cell kv : kvs) {
604      if (CellUtil.matchingFamily(kv, family) && CellUtil.matchingQualifier(kv, qualifier)) {
605        return true;
606      }
607    }
608    return false;
609  }
610
611  private Cell getColumn(final List<Cell> kvs, final byte[] family, final byte[] qualifier) {
612    for (Cell kv : kvs) {
613      if (CellUtil.matchingFamily(kv, family) && CellUtil.matchingQualifier(kv, qualifier)) {
614        return kv;
615      }
616    }
617    return null;
618  }
619
620  /** Use get to retrieve the HRegionInfo and validate it */
621  private void getRegionInfo(Table table) throws IOException {
622    Get get = new Get(ROW_KEY);
623    get.addColumn(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER);
624    Result result = table.get(get);
625    byte[] bytes = result.value();
626    validateRegionInfo(bytes);
627  }
628
629  /**
630   * Tests to do a sync flush during the middle of a scan. This is testing the StoreScanner update
631   * readers code essentially. This is not highly concurrent, since its all 1 thread. HBase-910.
632   */
633  @Test
634  public void testScanAndSyncFlush() throws Exception {
635    this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null);
636    Table hri = new RegionAsTable(region);
637    try {
638      LOG.info("Added: " + HTestConst.addContent(hri, Bytes.toString(HConstants.CATALOG_FAMILY),
639        Bytes.toString(HConstants.REGIONINFO_QUALIFIER)));
640      int count = count(hri, -1, false);
641      assertEquals(count, count(hri, 100, false)); // do a sync flush.
642    } catch (Exception e) {
643      LOG.error("Failed", e);
644      throw e;
645    } finally {
646      HBaseTestingUtil.closeRegionAndWAL(this.region);
647    }
648  }
649
650  /**
651   * Tests to do a concurrent flush (using a 2nd thread) while scanning. This tests both the
652   * StoreScanner update readers and the transition from memstore -> snapshot -> store file.
653   */
654  @Test
655  public void testScanAndRealConcurrentFlush() throws Exception {
656    this.region = TEST_UTIL.createLocalHRegion(TESTTABLEDESC, null, null);
657    Table hri = new RegionAsTable(region);
658    try {
659      LOG.info("Added: " + HTestConst.addContent(hri, Bytes.toString(HConstants.CATALOG_FAMILY),
660        Bytes.toString(HConstants.REGIONINFO_QUALIFIER)));
661      int count = count(hri, -1, false);
662      assertEquals(count, count(hri, 100, true)); // do a true concurrent background thread flush
663    } catch (Exception e) {
664      LOG.error("Failed", e);
665      throw e;
666    } finally {
667      HBaseTestingUtil.closeRegionAndWAL(this.region);
668    }
669  }
670
671  /**
672   * Make sure scanner returns correct result when we run a major compaction with deletes.
673   */
674  @Test
675  public void testScanAndConcurrentMajorCompact(TestInfo testInfo) throws Exception {
676    TableDescriptor htd =
677      TEST_UTIL.createTableDescriptor(TableName.valueOf(testInfo.getTestMethod().get().getName()),
678        ColumnFamilyDescriptorBuilder.DEFAULT_MIN_VERSIONS, 3, HConstants.FOREVER,
679        ColumnFamilyDescriptorBuilder.DEFAULT_KEEP_DELETED);
680    this.region = TEST_UTIL.createLocalHRegion(htd, null, null);
681    Table hri = new RegionAsTable(region);
682
683    try {
684      HTestConst.addContent(hri, Bytes.toString(fam1), Bytes.toString(col1), firstRowBytes,
685        secondRowBytes);
686      HTestConst.addContent(hri, Bytes.toString(fam2), Bytes.toString(col1), firstRowBytes,
687        secondRowBytes);
688
689      Delete dc = new Delete(firstRowBytes);
690      /* delete column1 of firstRow */
691      dc.addColumns(fam1, col1);
692      region.delete(dc);
693      region.flush(true);
694
695      HTestConst.addContent(hri, Bytes.toString(fam1), Bytes.toString(col1), secondRowBytes,
696        thirdRowBytes);
697      HTestConst.addContent(hri, Bytes.toString(fam2), Bytes.toString(col1), secondRowBytes,
698        thirdRowBytes);
699      region.flush(true);
700
701      InternalScanner s = region.getScanner(new Scan());
702      // run a major compact, column1 of firstRow will be cleaned.
703      region.compact(true);
704
705      List<Cell> results = new ArrayList<>();
706      s.next(results);
707
708      // make sure returns column2 of firstRow
709      assertEquals(1, results.size(), "result is not correct, keyValues : " + results);
710      assertTrue(CellUtil.matchingRows(results.get(0), firstRowBytes));
711      assertTrue(CellUtil.matchingFamily(results.get(0), fam2));
712
713      results = new ArrayList<>();
714      s.next(results);
715
716      // get secondRow
717      assertEquals(2, results.size());
718      assertTrue(CellUtil.matchingRows(results.get(0), secondRowBytes));
719      assertTrue(CellUtil.matchingFamily(results.get(0), fam1));
720      assertTrue(CellUtil.matchingFamily(results.get(1), fam2));
721    } finally {
722      HBaseTestingUtil.closeRegionAndWAL(this.region);
723    }
724  }
725
726  /**
727   * Count table.
728   * @param countTable Table
729   * @param flushIndex At what row we start the flush.
730   * @param concurrent if the flush should be concurrent or sync.
731   * @return Count of rows found.
732   */
733  private int count(final Table countTable, final int flushIndex, boolean concurrent)
734    throws Exception {
735    LOG.info("Taking out counting scan");
736    Scan scan = new Scan();
737    for (byte[] qualifier : EXPLICIT_COLS) {
738      scan.addColumn(HConstants.CATALOG_FAMILY, qualifier);
739    }
740    ResultScanner s = countTable.getScanner(scan);
741    int count = 0;
742    boolean justFlushed = false;
743    while (s.next() != null) {
744      if (justFlushed) {
745        LOG.info("after next() just after next flush");
746        justFlushed = false;
747      }
748      count++;
749      if (flushIndex == count) {
750        LOG.info("Starting flush at flush index " + flushIndex);
751        Thread t = new Thread() {
752          @Override
753          public void run() {
754            try {
755              region.flush(true);
756              LOG.info("Finishing flush");
757            } catch (IOException e) {
758              LOG.info("Failed flush cache");
759            }
760          }
761        };
762        t.start();
763        if (!concurrent) {
764          // sync flush
765          t.join();
766        }
767        LOG.info("Continuing on after kicking off background flush");
768        justFlushed = true;
769      }
770    }
771    s.close();
772    LOG.info("Found " + count + " items");
773    return count;
774  }
775}