Github user ohadshacham commented on a diff in the pull request:
https://github.com/apache/incubator-omid/pull/41#discussion_r206848753
--- Diff:
hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidSnapshotFilter.java
---
@@ -83,92 +76,97 @@ public void start(CoprocessorEnvironment env) throws
IOException {
if (commitTableName != null) {
commitTableConf.setTableName(commitTableName);
}
- if (commitTableClient == null) {
- commitTableClient = initAndGetCommitTableClient();
- }
- snapshotFilter = new SnapshotFilterImpl(commitTableClient);
-
LOG.info("Snapshot filter started");
}
@Override
public void stop(CoprocessorEnvironment e) throws IOException {
LOG.info("Stopping snapshot filter coprocessor");
- commitTableClient.close();
+ if (snapshotFilterQueue != null) {
+ for (SnapshotFilter snapshotFilter: snapshotFilterQueue) {
+
((SnapshotFilterImpl)snapshotFilter).getCommitTableClient().close();
+ }
+ }
LOG.info("Snapshot filter stopped");
}
- public void setCommitTableClient(CommitTable.Client commitTableClient)
{
- this.commitTableClient = commitTableClient;
- this.snapshotFilter.setCommitTableClient(commitTableClient);
- }
@Override
- public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> c,
Get get, List<Cell> result) throws IOException {
-
- if (get.getAttribute(CellUtils.CLIENT_GET_ATTRIBUTE) == null)
return;
-
- try {
- get.setAttribute(CellUtils.CLIENT_GET_ATTRIBUTE, null);
- RegionAccessWrapper regionAccessWrapper = new
RegionAccessWrapper(HBaseShims.getRegionCoprocessorRegion(c.getEnvironment()));
- Result res = regionAccessWrapper.get(get); // get parameters
were set at the client side
-
- snapshotFilter.setTableAccessWrapper(regionAccessWrapper);
+ public void postGetOp(ObserverContext<RegionCoprocessorEnvironment> e,
Get get, List<Cell> results)
+ throws IOException {
+ if (get.getFilter() != null) {
+ //This get had a filter and used a commit table client that
must put back
+ assert (get.getFilter() instanceof
TransactionVisibilityFilter);
+ TransactionVisibilityFilter filter =
(TransactionVisibilityFilter)get.getFilter();
+
snapshotFilterQueue.add((SnapshotFilterImpl)filter.getSnapshotFilter());
+ }
+ }
- List<Cell> filteredKeyValues = Collections.emptyList();
- if (!res.isEmpty()) {
- TSOProto.Transaction transaction =
TSOProto.Transaction.parseFrom(get.getAttribute(CellUtils.TRANSACTION_ATTRIBUTE));
- long id = transaction.getTimestamp();
- long readTs = transaction.getReadTimestamp();
- long epoch = transaction.getEpoch();
- VisibilityLevel visibilityLevel =
VisibilityLevel.fromInteger(transaction.getVisibilityLevel());
+ @Override
+ public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e,
Get get, List<Cell> results)
+ throws IOException {
+ if (get.getAttribute(CellUtils.CLIENT_GET_ATTRIBUTE) == null)
return;
- HBaseTransaction hbaseTransaction = new
HBaseTransaction(id, readTs, visibilityLevel, epoch, new
HashSet<HBaseCellId>(), new HashSet<HBaseCellId>(), null);
- filteredKeyValues =
snapshotFilter.filterCellsForSnapshot(res.listCells(), hbaseTransaction,
get.getMaxVersions(), new HashMap<String, Long>(), get.getAttributesMap());
- }
+ HBaseTransaction hbaseTransaction =
getHBaseTransaction(get.getAttribute(CellUtils.TRANSACTION_ATTRIBUTE));
+ SnapshotFilterImpl snapshotFilter = getSnapshotFilter(e);
+ get.setMaxVersions();
--- End diff --
I would add a comment that we set to max versions since we are doing Omid
filtering in the VisibilityFilter
---
