ThemisScanner extends AbstractClientScanner { public Result next() throws IOException { long beginTs = System.nanoTime(); Result pResult = null; boolean lockClean = false; try { pResult = this.scanner.next(); if (pResult == null) { return null; } if (ThemisCpUtil.isLockResult(pResult)) { lockClean = true; Get rowGet = createGetFromScan(scan, pResult.getRow()); pResult = transaction.tryToCleanLockAndGetAgain(tableName, rowGet, pResult.list()); if (pResult.isEmpty()) { return next(); } else { return pResult; } } return pResult; } finally { ThemisStatistics.updateLatency(ThemisStatistics.getStatistics().nextLatency, beginTs); ThemisStatistics.logSlowOperation("themisNext", beginTs, "row=" + pResult + ", lockClean=" + lockClean); } } ThemisScanner(final byte[] tableName, final Scan scan, final Transaction transaction); static Get createGetFromScan(Scan scan, byte[] rowkey); Result next(); void close(); @Override Result[] next(int nbRows);  }
@Test public void testScanOneRow() throws Exception { ColumnCoordinate[] columns = new ColumnCoordinate[]{COLUMN}; ThemisScanner scanner = prepareScanner(columns); checkAndCloseScanner(scanner); columns = new ColumnCoordinate[]{COLUMN}; prepareScanData(columns); scanner = prepareScanner(columns); Result result = scanner.next(); checkScanRow(columns, result); checkAndCloseScanner(scanner); columns = new ColumnCoordinate[]{COLUMN, COLUMN_WITH_ANOTHER_FAMILY, COLUMN_WITH_ANOTHER_QUALIFIER}; prepareScanData(columns); scanner = prepareScanner(columns); result = scanner.next(); checkScanRow(columns, result); checkAndCloseScanner(scanner); scanner = transaction.getScanner(TABLENAME, new ThemisScan().addFamily(FAMILY)); result = scanner.next(); checkScanRow(new ColumnCoordinate[]{COLUMN, COLUMN_WITH_ANOTHER_QUALIFIER}, result); checkAndCloseScanner(scanner); scanner = transaction.getScanner(TABLENAME, new ThemisScan()); result = scanner.next(); checkScanRow(columns, result); checkAndCloseScanner(scanner); scanner = transaction.getScanner(TABLENAME, new ThemisScan().addFamily(ANOTHER_FAMILY) .addColumn(FAMILY, QUALIFIER)); result = scanner.next(); checkScanRow(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_FAMILY, COLUMN}, result); checkAndCloseScanner(scanner); }
@Test public void testScanMultiRows() throws Exception { prepareScanData(TRANSACTION_COLUMNS); ThemisScanner scanner = prepareScanner(TRANSACTION_COLUMNS); Result result = scanner.next(); checkScanRow(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_ROW}, result); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); ThemisScan pScan = prepareScan(TRANSACTION_COLUMNS); pScan.setStartRow(ROW).setStopRow(ROW); scanner = transaction.getScanner(TABLENAME, pScan); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); }
@Test public void testScanWithDeletedRow() throws Exception { nextTransactionTs(); writePutAndData(COLUMN, prewriteTs - 6, commitTs - 6); createTransactionWithMock(); ThemisScanner scanner = prepareScanner(new ColumnCoordinate[]{COLUMN}); checkReadColumnResultWithTs(scanner.next(), COLUMN, prewriteTs - 6); checkAndCloseScanner(scanner); writeDeleteColumn(COLUMN, prewriteTs - 3, commitTs - 3); scanner = prepareScanner(new ColumnCoordinate[]{COLUMN}); checkAndCloseScanner(scanner); writeDeleteColumn(COLUMN_WITH_ANOTHER_ROW, prewriteTs - 2, commitTs - 2); writePutAndData(COLUMN, prewriteTs - 2, commitTs - 2); writePutAndData(COLUMN_WITH_ZZ_ROW, prewriteTs - 2, commitTs - 2); scanner = prepareScanner(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_ROW, COLUMN, COLUMN_WITH_ZZ_ROW}); checkReadColumnResultWithTs(scanner.next(), COLUMN, prewriteTs - 2); checkReadColumnResultWithTs(scanner.next(), COLUMN_WITH_ZZ_ROW, prewriteTs - 2); checkAndCloseScanner(scanner); }
@Test public void testScanWithDelete() throws Exception { ColumnCoordinate[] scanColumns = new ColumnCoordinate[]{COLUMN, COLUMN_WITH_ANOTHER_FAMILY, COLUMN_WITH_ANOTHER_QUALIFIER}; nextTransactionTs(); createTransactionWithMock(); writeDeleteAfterPut(COLUMN_WITH_ANOTHER_FAMILY, prewriteTs - 5, commitTs - 5); writePutAfterDelete(COLUMN_WITH_ANOTHER_QUALIFIER, prewriteTs - 5, commitTs - 5); ThemisScanner scanner = prepareScanner(scanColumns); Result result = scanner.next(); checkReadColumnResultWithTs(result, COLUMN_WITH_ANOTHER_QUALIFIER, prewriteTs - 5); checkAndCloseScanner(scanner); writeDeleteAfterPut(COLUMN_WITH_ANOTHER_FAMILY, prewriteTs + 1, commitTs + 1); writePutAfterDelete(COLUMN_WITH_ANOTHER_QUALIFIER, prewriteTs + 1, commitTs + 1); scanner = prepareScanner(scanColumns); result = scanner.next(); checkReadColumnResultWithTs(result, COLUMN_WITH_ANOTHER_FAMILY, prewriteTs - 2); checkAndCloseScanner(scanner); }
@Test public void testScanWithEmptyRowLeftAfterLockClean() throws Exception { prepareScanData(new ColumnCoordinate[]{}); writeLockAndData(COLUMN_WITH_ANOTHER_ROW, lastTs(prewriteTs)); Mockito.when(mockRegister.isWorkerAlive(TestBase.CLIENT_TEST_ADDRESS)).thenReturn(false); ThemisScan pScan = prepareScan(TRANSACTION_COLUMNS); pScan.setCaching(10); ThemisScanner scanner = transaction.getScanner(TABLENAME, pScan); checkAndCloseScanner(scanner); prepareScanData(new ColumnCoordinate[]{COLUMN, COLUMN_WITH_ANOTHER_FAMILY, COLUMN_WITH_ANOTHER_QUALIFIER}); writeLockAndData(COLUMN_WITH_ANOTHER_ROW, lastTs(prewriteTs)); Mockito.when(mockRegister.isWorkerAlive(TestBase.CLIENT_TEST_ADDRESS)).thenReturn(false); pScan = prepareScan(TRANSACTION_COLUMNS); pScan.setCaching(10); scanner = transaction.getScanner(TABLENAME, pScan); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); }
@Test public void testScanWithRowkeyFilter() throws Exception { prepareScanData(TRANSACTION_COLUMNS); ThemisScanner scanner = prepareScanner(TRANSACTION_COLUMNS, new PrefixFilter(ROW)); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); scanner = prepareScanner(TRANSACTION_COLUMNS, new PrefixFilter(ANOTHER_ROW)); checkScanRow(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_ROW}, scanner.next()); checkAndCloseScanner(scanner); FilterList filterList = new FilterList(); filterList.addFilter(new PrefixFilter(ROW)); filterList.addFilter(new PrefixFilter(ANOTHER_ROW)); scanner = prepareScanner(TRANSACTION_COLUMNS, filterList); checkAndCloseScanner(scanner); filterList = new FilterList(Operator.MUST_PASS_ONE); filterList.addFilter(new PrefixFilter(ROW)); filterList.addFilter(new PrefixFilter(ANOTHER_ROW)); scanner = prepareScanner(TRANSACTION_COLUMNS, filterList); checkScanRow(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_ROW}, scanner.next()); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); }
@Test public void testScanWithNoRowkeyFilter() throws Exception { prepareScanData(TRANSACTION_COLUMNS); ValueFilter valueFilter = new ValueFilter(CompareOp.EQUAL, new BinaryComparator(ANOTHER_VALUE)); ThemisScanner scanner = prepareScanner(TRANSACTION_COLUMNS, valueFilter); checkAndCloseScanner(scanner); SingleColumnValueFilter singleColumnFilter = new SingleColumnValueFilter(ANOTHER_FAMILY, QUALIFIER, CompareOp.EQUAL, VALUE); singleColumnFilter.setFilterIfMissing(true); scanner = prepareScanner(TRANSACTION_COLUMNS, singleColumnFilter); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); valueFilter = new ValueFilter(CompareOp.EQUAL, new BinaryComparator(VALUE)); singleColumnFilter = new SingleColumnValueFilter(QUALIFIER, QUALIFIER, CompareOp.EQUAL, ANOTHER_VALUE); singleColumnFilter.setFilterIfMissing(true); FilterList filterList = new FilterList(); filterList.addFilter(valueFilter); filterList.addFilter(singleColumnFilter); scanner = prepareScanner(TRANSACTION_COLUMNS, filterList); checkAndCloseScanner(scanner); filterList = new FilterList(Operator.MUST_PASS_ONE); filterList.addFilter(valueFilter); filterList.addFilter(singleColumnFilter); scanner = prepareScanner(TRANSACTION_COLUMNS, filterList); checkScanRow(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_ROW}, scanner.next()); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); }
@Test public void testScanWithCustomerFilters() throws IOException { prepareScanData(TRANSACTION_COLUMNS); CustomerRowkeyFilter rowLevelFilter = new CustomerRowkeyFilter(ROW); ThemisScanner scanner = prepareScanner(TRANSACTION_COLUMNS, rowLevelFilter); checkResultForROW(scanner.next()); checkAndCloseScanner(scanner); CustomerColumnFilter columnFilter = new CustomerColumnFilter(QUALIFIER); scanner = prepareScanner(TRANSACTION_COLUMNS, columnFilter); checkScanRow(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_ROW}, scanner.next()); checkScanRow(new ColumnCoordinate[]{COLUMN, COLUMN_WITH_ANOTHER_FAMILY}, scanner.next()); checkAndCloseScanner(scanner); }
@Test public void testScanWithFilter() throws IOException { prepareScanData(TRANSACTION_COLUMNS); writeData(COLUMN, lastTs(prewriteTs), ANOTHER_VALUE); ValueFilter valueFilter = new ValueFilter(CompareOp.EQUAL, new BinaryComparator(ANOTHER_VALUE)); PrefixFilter prefixFilter = new PrefixFilter(ANOTHER_ROW); FilterList filterList = new FilterList(); filterList.addFilter(valueFilter); filterList.addFilter(prefixFilter); ThemisScanner scanner = prepareScanner(TRANSACTION_COLUMNS, filterList); checkAndCloseScanner(scanner); filterList = new FilterList(Operator.MUST_PASS_ONE); filterList.addFilter(valueFilter); filterList.addFilter(prefixFilter); scanner = prepareScanner(TRANSACTION_COLUMNS, filterList); checkScanRow(new ColumnCoordinate[]{COLUMN_WITH_ANOTHER_ROW}, scanner.next()); Assert.assertEquals(1, scanner.next().size()); checkAndCloseScanner(scanner); }