diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java index d3a1e49d8e..b22cd5bd69 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java @@ -142,6 +142,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner private ReentrantLock flushLock = new ReentrantLock(); private final long readPt; + private boolean topChanged = false; // used by the injection framework to test race between StoreScanner construction and compaction enum StoreScannerCompactionRace { @@ -547,7 +548,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner if (prevCell != cell) ++kvsScanned; // Do object compare - we set prevKV from the same heap. checkScanOrder(prevCell, cell, comparator); prevCell = cell; - + topChanged = false; ScanQueryMatcher.MatchCode qcode = matcher.match(cell); switch (qcode) { case INCLUDE: @@ -648,6 +649,16 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner case SEEK_NEXT_COL: seekOrSkipToNextColumn(cell); + // If the top cell won't be flushed into disk, + // the new top cell may be changed after #reopenAfterFlush. + // Because the older top cell only exist in the memstore scanner but the memstore + // scanner is replaced by hfile scanner after #reopenAfterFlush. + // If the row of top cell is changed, we should return the current cells. + // Otherwise, we may return the cells across different rows. + if (!outResult.isEmpty() && topChanged) { + return scannerContext.setScannerState(heap.peek() == null ? + NextState.NO_MORE_VALUES : NextState.MORE_VALUES).hasMoreValues(); + } break; case SKIP: @@ -846,14 +857,16 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner resetScannerStack(this.lastTop); if (this.heap.peek() == null || store.getComparator().compareRows(this.lastTop, this.heap.peek()) != 0) { - LOG.debug("Storescanner.peek() is changed where before = " + LOG.info("Storescanner.peek() is changed where before = " + this.lastTop.toString() + ",and after = " + this.heap.peek()); this.lastTop = null; + topChanged = true; return true; } this.lastTop = null; // gone! } // else dont need to reseek + topChanged = false; return false; } diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java index 573de11ffb..36ca6031b8 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java @@ -178,7 +178,7 @@ public class TestStore { @SuppressWarnings("deprecation") private Store init(String methodName, Configuration conf, HTableDescriptor htd, - HColumnDescriptor hcd, MyScannerHook hook) throws IOException { + HColumnDescriptor hcd, MyStoreHook hook) throws IOException { //Setting up a Store Path basedir = new Path(DIR+methodName); Path tableDir = FSUtils.getTableDir(basedir, htd.getTableName()); @@ -1154,6 +1154,10 @@ public class TestStore { } private Cell createCell(byte[] qualifier, long ts, long sequenceId, byte[] value) throws IOException { + return createCell(row, qualifier, ts, sequenceId, value); + } + + private Cell createCell(byte[] row, byte[] qualifier, long ts, long sequenceId, byte[] value) throws IOException { Cell c = CellUtil.createCell(row, family, qualifier, ts, KeyValue.Type.Put.getCode(), value); CellUtil.setSequenceId(c, sequenceId); return c; @@ -1163,9 +1167,9 @@ public class TestStore { public void testScanWithDoubleFlush() throws IOException { Configuration conf = HBaseConfiguration.create(); // Initialize region - MyStore myStore = initMyStore(name.getMethodName(), conf, new MyScannerHook() { + MyStore myStore = initMyStore(name.getMethodName(), conf, new MyStoreHook() { @Override - public void hook(final MyStore store) throws IOException { + public void getScanners(final MyStore store) throws IOException { final long tmpId = id++; ExecutorService s = Executors.newSingleThreadExecutor(); s.submit(new Runnable() { @@ -1231,6 +1235,77 @@ public class TestStore { } @Test + public void testFlushBeforeCompletingScan() throws IOException, InterruptedException { + Configuration conf = HBaseConfiguration.create(); + HColumnDescriptor hcd = new HColumnDescriptor(family); + hcd.setMaxVersions(1); + byte[] r0 = Bytes.toBytes("row0"); + byte[] r1 = Bytes.toBytes("row1"); + byte[] r2 = Bytes.toBytes("row2"); + byte[] value0 = Bytes.toBytes("value0"); + byte[] value1 = Bytes.toBytes("value1"); + byte[] value2 = Bytes.toBytes("value2"); + long ts = EnvironmentEdgeManager.currentTime(); + final long seqId = 100; + init(name.getMethodName(), conf, new HTableDescriptor(TableName.valueOf(table)), hcd, new MyStoreHook() { + @Override + public long getSmallestReadPoint(HStore store) { + return seqId + 3; + } + }); + // The cells having the value0 won't be flushed to disk because the value of max version is 1 + store.add(createCell(r0, qf1, ts, seqId, value0)); + store.add(createCell(r0, qf2, ts, seqId, value0)); + store.add(createCell(r0, qf3, ts, seqId, value0)); + store.add(createCell(r1, qf1, ts + 1, seqId + 1, value1)); + store.add(createCell(r1, qf2, ts + 1, seqId + 1, value1)); + store.add(createCell(r1, qf3, ts + 1, seqId + 1, value1)); + store.add(createCell(r2, qf1, ts + 2, seqId + 2, value2)); + store.add(createCell(r2, qf2, ts + 2, seqId + 2, value2)); + store.add(createCell(r2, qf3, ts + 2, seqId + 2, value2)); + store.add(createCell(r1, qf1, ts + 3, seqId + 3, value1)); + store.add(createCell(r1, qf2, ts + 3, seqId + 3, value1)); + store.add(createCell(r1, qf3, ts + 3, seqId + 3, value1)); + InternalScanner scanner = (InternalScanner) store.getScanner( + new Scan().withStartRow(r1), null, seqId + 3); + List myList = new MyList<>(new MyListHook() { + @Override + public void hook(int size) { + if (size == 2) { + try { + flushStore(store, id++); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + }); + try { + // r1 + scanner.next(myList); + assertEquals(3, myList.size()); + for (Cell c : myList) { + byte[] actualValue = CellUtil.cloneValue(c); + assertTrue("expected:" + Bytes.toStringBinary(value1) + + ", actual:" + Bytes.toStringBinary(actualValue) + , Bytes.equals(actualValue, value1)); + } + List normalList = new ArrayList<>(3); + // r2 + scanner.next(normalList); + assertEquals(3, normalList.size()); + for (Cell c : normalList) { + byte[] actualValue = CellUtil.cloneValue(c); + assertTrue("expected:" + Bytes.toStringBinary(value2) + + ", actual:" + Bytes.toStringBinary(actualValue) + , Bytes.equals(actualValue, value2)); + } + } finally { + scanner.close(); + } + } + + @Test public void testReclaimChunkWhenScaning() throws IOException { Configuration conf = HBaseConfiguration.create(); conf.setFloat(CHUNK_POOL_MAXSIZE_KEY, 1); @@ -1291,7 +1366,7 @@ public class TestStore { } } - private MyStore initMyStore(String methodName, Configuration conf, MyScannerHook hook) throws IOException { + private MyStore initMyStore(String methodName, Configuration conf, MyStoreHook hook) throws IOException { HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(table)); HColumnDescriptor hcd = new HColumnDescriptor(family); hcd.setMaxVersions(5); @@ -1300,10 +1375,10 @@ public class TestStore { private static class MyStore extends HStore { - private final MyScannerHook hook; + private final MyStoreHook hook; MyStore(final HRegion region, final HColumnDescriptor family, - final Configuration confParam, MyScannerHook hook) throws IOException { + final Configuration confParam, MyStoreHook hook) throws IOException { super(region, family, confParam); this.hook = hook; } @@ -1312,15 +1387,23 @@ public class TestStore { public List getScanners(List files, boolean cacheBlocks, boolean isGet, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher, byte[] startRow, byte[] stopRow, long readPt, boolean includeMemstoreScanner) throws IOException { - hook.hook(this); + hook.getScanners(this); return super.getScanners(files, cacheBlocks, isGet, usePread, isCompaction, matcher, startRow, stopRow, readPt, includeMemstoreScanner); } - } - private interface MyScannerHook { + @Override + public long getSmallestReadPoint() { + return hook.getSmallestReadPoint(this); + } + } - void hook(MyStore store) throws IOException; + private abstract class MyStoreHook { + void getScanners(MyStore store) throws IOException { + } + long getSmallestReadPoint(HStore store) { + return store.getHRegion().getSmallestReadPoint(); + } } interface MyListHook {