TransactionVisibilityFilter extends FilterBase { @Override public ReturnCode filterKeyValue(Cell cell) throws IOException { if (!CellUtil.matchingFamily(cell, currentFamily)) { currentFamily = CellUtil.cloneFamily(cell); Long familyOldestTs = oldestTsByFamily.get(currentFamily); currentOldestTs = familyOldestTs != null ? familyOldestTs : 0; deleteTracker.reset(); } long kvTimestamp = cell.getTimestamp(); if (TxUtils.getTimestampForTTL(kvTimestamp) < currentOldestTs) { return ReturnCode.NEXT_COL; } else if (tx.isVisible(kvTimestamp)) { if (tx.getVisibilityLevel() == Transaction.VisibilityLevel.SNAPSHOT_ALL && tx.isCurrentWrite(kvTimestamp)) { return runSubFilter(ReturnCode.INCLUDE, cell); } if (DeleteTracker.isFamilyDelete(cell)) { deleteTracker.addFamilyDelete(cell); if (clearDeletes) { return ReturnCode.NEXT_COL; } else { return runSubFilter(ReturnCode.INCLUDE_AND_NEXT_COL, cell); } } if (deleteTracker.isDeleted(cell)) { return ReturnCode.NEXT_COL; } if (isColumnDelete(cell)) { if (clearDeletes) { return ReturnCode.NEXT_COL; } else { return runSubFilter(ReturnCode.INCLUDE_AND_NEXT_COL, cell); } } return runSubFilter(ReturnCode.INCLUDE_AND_NEXT_COL, cell); } else { return ReturnCode.SKIP; } } TransactionVisibilityFilter(Transaction tx, Map<byte[], Long> ttlByFamily, boolean allowEmptyValues,
                                     ScanType scanType);  TransactionVisibilityFilter(Transaction tx, Map<byte[], Long> ttlByFamily, boolean allowEmptyValues,
                               ScanType scanType, @Nullable Filter cellFilter); @Override ReturnCode filterKeyValue(Cell cell); @Override boolean filterRow(); @Override Cell transformCell(Cell cell); @Override void reset(); @Override boolean filterRowKey(byte[] buffer, int offset, int length); @Override boolean filterAllRemaining(); @Override void filterRowCells(List<Cell> kvs); @Override boolean hasFilterRow(); @SuppressWarnings("deprecation") @Override KeyValue getNextKeyHint(KeyValue currentKV); @Override Cell getNextCellHint(Cell currentKV); @Override boolean isFamilyEssential(byte[] name); @Override byte[] toByteArray();  }
@Test public void testTTLFiltering() throws Exception { Map<byte[], Long> ttls = Maps.newTreeMap(Bytes.BYTES_COMPARATOR); ttls.put(FAM, 10L); ttls.put(FAM2, 30L); ttls.put(FAM3, 0L); Transaction tx = txManager.startShort(); long now = tx.getVisibilityUpperBound(); Filter filter = new TransactionVisibilityFilter(tx, ttls, false, ScanType.USER_SCAN); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM, "val1", now))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM, "val1", now - 1 * TxConstants.MAX_TX_PER_MS))); assertEquals(Filter.ReturnCode.NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM, "val1", now - 11 * TxConstants.MAX_TX_PER_MS))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM2, "val1", now - 11 * TxConstants.MAX_TX_PER_MS))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM2, "val1", now - 21 * TxConstants.MAX_TX_PER_MS))); assertEquals(Filter.ReturnCode.NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM2, "val1", now - 31 * TxConstants.MAX_TX_PER_MS))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", now - 31 * TxConstants.MAX_TX_PER_MS))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", now - 1001 * TxConstants.MAX_TX_PER_MS))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row2", FAM, "val1", now))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row2", FAM, "val1", now - 1 * TxConstants.MAX_TX_PER_MS))); long preNow = now / TxConstants.MAX_TX_PER_MS; assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow - 9L))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow - 10L))); assertEquals(Filter.ReturnCode.NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow - 11L))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow - 9L))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow - 10L))); assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL, filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow - 1001L))); }