hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionVisibilityFilterBase.java - phoenix-omid - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.omid.transaction;

 import org.apache.phoenix.thirdparty.com.google.common.base.Optional;


 import org.apache.commons.collections4.map.LRUMap;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.filter.Filter;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.omid.OmidFilterBase;


 import java.io.IOException;
 import java.util.HashMap;

 import java.util.List;
 import java.util.Map;


 public class TransactionVisibilityFilterBase extends OmidFilterBase {

     // optional sub-filter to apply to visible cells
     private final Filter userFilter;
     private final SnapshotFilterImpl snapshotFilter;
     private final LRUMap<Long ,Long> commitCache;
     private final HBaseTransaction hbaseTransaction;

     // This cache is cleared when moving to the next row
     // So no need to keep row name
     private final Map<ImmutableBytesWritable, Long> familyDeletionCache;

     public TransactionVisibilityFilterBase(Filter cellFilter,
                                            SnapshotFilterImpl snapshotFilter,
                                            HBaseTransaction hbaseTransaction) {
         this.userFilter = cellFilter;
         this.snapshotFilter = snapshotFilter;
         commitCache = new LRUMap<>(1000);
         this.hbaseTransaction = hbaseTransaction;
         familyDeletionCache = new HashMap<>();

     }

     @Override
     public ReturnCode filterKeyValue(Cell v) throws IOException {
         if (CellUtils.isShadowCell(v)) {
             Long commitTs =  Bytes.toLong(CellUtil.cloneValue(v));
             commitCache.put(v.getTimestamp(), commitTs);
             // Continue getting shadow cells until one of them fits this transaction
             if (hbaseTransaction.getStartTimestamp() >= commitTs) {
                 return ReturnCode.NEXT_COL;
             } else {
                 return ReturnCode.SKIP;
             }
         }

         Optional<Long> commitTS = getCommitIfInSnapshot(v, CellUtils.isFamilyDeleteCell(v));
         if (commitTS.isPresent()) {
             if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL &&
                     snapshotFilter.getTSIfInTransaction(v, hbaseTransaction).isPresent()) {
                 return runUserFilter(v, ReturnCode.INCLUDE);
             }
             if (CellUtils.isFamilyDeleteCell(v)) {
                 familyDeletionCache.put(createImmutableBytesWritable(v), commitTS.get());
                 if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL) {
                     return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
                 } else {
                     return ReturnCode.NEXT_COL;
                 }
             }
             Long deleteCommit = familyDeletionCache.get(createImmutableBytesWritable(v));
             if (deleteCommit != null && deleteCommit >= v.getTimestamp()) {
                 if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL) {
                     return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
                 } else {
                     return ReturnCode.NEXT_COL;
                 }
             }
             if (CellUtils.isTombstone(v)) {
                 if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL) {
                     return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
                 } else {
                     return ReturnCode.NEXT_COL;
                 }
             }

             return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
         }

         return ReturnCode.SKIP;
     }


     private ImmutableBytesWritable createImmutableBytesWritable(Cell v) {
         return new ImmutableBytesWritable(v.getFamilyArray(),
                 v.getFamilyOffset(),v.getFamilyLength());
     }

     private ReturnCode runUserFilter(Cell v, ReturnCode snapshotReturn)
             throws IOException {
         assert(snapshotReturn == ReturnCode.INCLUDE_AND_NEXT_COL || snapshotReturn == ReturnCode.INCLUDE);
         if (userFilter == null) {
             return snapshotReturn;
         }

         ReturnCode userRes = userFilter.filterKeyValue(v);
         switch (userRes) {
             case INCLUDE:
                 return snapshotReturn;
             case SKIP:
                 return (snapshotReturn == ReturnCode.INCLUDE) ? ReturnCode.SKIP: ReturnCode.NEXT_COL;
             default:
                 return userRes;
         }

     }

     // For family delete cells, the sc hasn't arrived yet so get sc from region before going to ct
     private Optional<Long> getCommitIfInSnapshot(Cell v, boolean getShadowCellBeforeCT) throws IOException {
         Long cachedCommitTS = commitCache.get(v.getTimestamp());
         if (cachedCommitTS != null) {
             if (hbaseTransaction.getStartTimestamp() >= cachedCommitTS) {
                 return Optional.of(cachedCommitTS);
             } else {
                 return Optional.absent();
             }
         }
         if (snapshotFilter.getTSIfInTransaction(v, hbaseTransaction).isPresent()) {
             return Optional.of(v.getTimestamp());
         }

         if (getShadowCellBeforeCT) {

             // Try to get shadow cell from region
             final Get get = new Get(CellUtil.cloneRow(v));
             get.setTimeStamp(v.getTimestamp()).setMaxVersions(1);
             get.addColumn(CellUtil.cloneFamily(v), CellUtils.addShadowCellSuffixPrefix(CellUtils.FAMILY_DELETE_QUALIFIER));
             Result shadowCell = snapshotFilter.getTableAccessWrapper().get(get);

             if (!shadowCell.isEmpty()) {
                 long commitTS = Bytes.toLong(CellUtil.cloneValue(shadowCell.rawCells()[0]));
                 commitCache.put(v.getTimestamp(), commitTS);
                 if (commitTS <= hbaseTransaction.getStartTimestamp()) {
                     return Optional.of(commitTS);
                 }
             }
         }

         Optional<Long> commitTS = snapshotFilter.getTSIfInSnapshot(v, hbaseTransaction, commitCache);
         if (commitTS.isPresent()) {
             commitCache.put(v.getTimestamp(), commitTS.get());
         } else {
             commitCache.put(v.getTimestamp(), Long.MAX_VALUE);
         }
         return commitTS;
     }


     @Override
     public void reset() throws IOException {
         familyDeletionCache.clear();
         if (userFilter != null) {
             userFilter.reset();
         }
     }

     @Override
     public boolean filterRow() throws IOException {
         if (userFilter != null) {
             return userFilter.filterRow();
         }
         return super.filterRow();
     }


     @Override
     public boolean filterRowKey(byte[] buffer, int offset, int length) throws IOException {
         if (userFilter != null) {
             return userFilter.filterRowKey(buffer, offset, length);
         }
         return super.filterRowKey(buffer, offset, length);
     }

     @Override
     public boolean filterAllRemaining() throws IOException {
         if (userFilter != null) {
             return userFilter.filterAllRemaining();
         }
         return super.filterAllRemaining();
     }

     @Override
     public void filterRowCells(List<Cell> kvs) throws IOException {
         if (userFilter != null) {
             userFilter.filterRowCells(kvs);
         } else {
             super.filterRowCells(kvs);
         }
     }

     @Override
     public boolean hasFilterRow() {
         if (userFilter != null) {
             return userFilter.hasFilterRow();
         }
         return super.hasFilterRow();
     }

     @Override
     public Cell getNextCellHint(Cell currentKV) throws IOException {
         if (userFilter != null) {
             return userFilter.getNextCellHint(currentKV);
         }
         return super.getNextCellHint(currentKV);
     }

     @Override
     public boolean isFamilyEssential(byte[] name) throws IOException {
         if (userFilter != null) {
             return userFilter.isFamilyEssential(name);
         }
         return super.isFamilyEssential(name);
     }

     @Override
     public byte[] toByteArray() throws IOException {
         return super.toByteArray();
     }

     public Filter getInnerFilter() {
         return userFilter;
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.omid.transaction;

	import org.apache.phoenix.thirdparty.com.google.common.base.Optional;


	import org.apache.commons.collections4.map.LRUMap;
	import org.apache.hadoop.hbase.Cell;
	import org.apache.hadoop.hbase.CellUtil;
	import org.apache.hadoop.hbase.client.Get;
	import org.apache.hadoop.hbase.client.Result;
	import org.apache.hadoop.hbase.filter.Filter;
	import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

	import org.apache.hadoop.hbase.util.Bytes;
	import org.apache.omid.OmidFilterBase;


	import java.io.IOException;
	import java.util.HashMap;

	import java.util.List;
	import java.util.Map;


	public class TransactionVisibilityFilterBase extends OmidFilterBase {

	// optional sub-filter to apply to visible cells
	private final Filter userFilter;
	private final SnapshotFilterImpl snapshotFilter;
	private final LRUMap<Long ,Long> commitCache;
	private final HBaseTransaction hbaseTransaction;

	// This cache is cleared when moving to the next row
	// So no need to keep row name
	private final Map<ImmutableBytesWritable, Long> familyDeletionCache;

	public TransactionVisibilityFilterBase(Filter cellFilter,
	SnapshotFilterImpl snapshotFilter,
	HBaseTransaction hbaseTransaction) {
	this.userFilter = cellFilter;
	this.snapshotFilter = snapshotFilter;
	commitCache = new LRUMap<>(1000);
	this.hbaseTransaction = hbaseTransaction;
	familyDeletionCache = new HashMap<>();

	}

	@Override
	public ReturnCode filterKeyValue(Cell v) throws IOException {
	if (CellUtils.isShadowCell(v)) {
	Long commitTs = Bytes.toLong(CellUtil.cloneValue(v));
	commitCache.put(v.getTimestamp(), commitTs);
	// Continue getting shadow cells until one of them fits this transaction
	if (hbaseTransaction.getStartTimestamp() >= commitTs) {
	return ReturnCode.NEXT_COL;
	} else {
	return ReturnCode.SKIP;
	}
	}

	Optional<Long> commitTS = getCommitIfInSnapshot(v, CellUtils.isFamilyDeleteCell(v));
	if (commitTS.isPresent()) {
	if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL &&
	snapshotFilter.getTSIfInTransaction(v, hbaseTransaction).isPresent()) {
	return runUserFilter(v, ReturnCode.INCLUDE);
	}
	if (CellUtils.isFamilyDeleteCell(v)) {
	familyDeletionCache.put(createImmutableBytesWritable(v), commitTS.get());
	if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL) {
	return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
	} else {
	return ReturnCode.NEXT_COL;
	}
	}
	Long deleteCommit = familyDeletionCache.get(createImmutableBytesWritable(v));
	if (deleteCommit != null && deleteCommit >= v.getTimestamp()) {
	if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL) {
	return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
	} else {
	return ReturnCode.NEXT_COL;
	}
	}
	if (CellUtils.isTombstone(v)) {
	if (hbaseTransaction.getVisibilityLevel() == AbstractTransaction.VisibilityLevel.SNAPSHOT_ALL) {
	return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
	} else {
	return ReturnCode.NEXT_COL;
	}
	}

	return runUserFilter(v, ReturnCode.INCLUDE_AND_NEXT_COL);
	}

	return ReturnCode.SKIP;
	}


	private ImmutableBytesWritable createImmutableBytesWritable(Cell v) {
	return new ImmutableBytesWritable(v.getFamilyArray(),
	v.getFamilyOffset(),v.getFamilyLength());
	}

	private ReturnCode runUserFilter(Cell v, ReturnCode snapshotReturn)
	throws IOException {
	assert(snapshotReturn == ReturnCode.INCLUDE_AND_NEXT_COL \|\| snapshotReturn == ReturnCode.INCLUDE);
	if (userFilter == null) {
	return snapshotReturn;
	}

	ReturnCode userRes = userFilter.filterKeyValue(v);
	switch (userRes) {
	case INCLUDE:
	return snapshotReturn;
	case SKIP:
	return (snapshotReturn == ReturnCode.INCLUDE) ? ReturnCode.SKIP: ReturnCode.NEXT_COL;
	default:
	return userRes;
	}

	}

	// For family delete cells, the sc hasn't arrived yet so get sc from region before going to ct
	private Optional<Long> getCommitIfInSnapshot(Cell v, boolean getShadowCellBeforeCT) throws IOException {
	Long cachedCommitTS = commitCache.get(v.getTimestamp());
	if (cachedCommitTS != null) {
	if (hbaseTransaction.getStartTimestamp() >= cachedCommitTS) {
	return Optional.of(cachedCommitTS);
	} else {
	return Optional.absent();
	}
	}
	if (snapshotFilter.getTSIfInTransaction(v, hbaseTransaction).isPresent()) {
	return Optional.of(v.getTimestamp());
	}

	if (getShadowCellBeforeCT) {

	// Try to get shadow cell from region
	final Get get = new Get(CellUtil.cloneRow(v));
	get.setTimeStamp(v.getTimestamp()).setMaxVersions(1);
	get.addColumn(CellUtil.cloneFamily(v), CellUtils.addShadowCellSuffixPrefix(CellUtils.FAMILY_DELETE_QUALIFIER));
	Result shadowCell = snapshotFilter.getTableAccessWrapper().get(get);

	if (!shadowCell.isEmpty()) {
	long commitTS = Bytes.toLong(CellUtil.cloneValue(shadowCell.rawCells()[0]));
	commitCache.put(v.getTimestamp(), commitTS);
	if (commitTS <= hbaseTransaction.getStartTimestamp()) {
	return Optional.of(commitTS);
	}
	}
	}

	Optional<Long> commitTS = snapshotFilter.getTSIfInSnapshot(v, hbaseTransaction, commitCache);
	if (commitTS.isPresent()) {
	commitCache.put(v.getTimestamp(), commitTS.get());
	} else {
	commitCache.put(v.getTimestamp(), Long.MAX_VALUE);
	}
	return commitTS;
	}


	@Override
	public void reset() throws IOException {
	familyDeletionCache.clear();
	if (userFilter != null) {
	userFilter.reset();
	}
	}

	@Override
	public boolean filterRow() throws IOException {
	if (userFilter != null) {
	return userFilter.filterRow();
	}
	return super.filterRow();
	}


	@Override
	public boolean filterRowKey(byte[] buffer, int offset, int length) throws IOException {
	if (userFilter != null) {
	return userFilter.filterRowKey(buffer, offset, length);
	}
	return super.filterRowKey(buffer, offset, length);
	}

	@Override
	public boolean filterAllRemaining() throws IOException {
	if (userFilter != null) {
	return userFilter.filterAllRemaining();
	}
	return super.filterAllRemaining();
	}

	@Override
	public void filterRowCells(List<Cell> kvs) throws IOException {
	if (userFilter != null) {
	userFilter.filterRowCells(kvs);
	} else {
	super.filterRowCells(kvs);
	}
	}

	@Override
	public boolean hasFilterRow() {
	if (userFilter != null) {
	return userFilter.hasFilterRow();
	}
	return super.hasFilterRow();
	}

	@Override
	public Cell getNextCellHint(Cell currentKV) throws IOException {
	if (userFilter != null) {
	return userFilter.getNextCellHint(currentKV);
	}
	return super.getNextCellHint(currentKV);
	}

	@Override
	public boolean isFamilyEssential(byte[] name) throws IOException {
	if (userFilter != null) {
	return userFilter.isFamilyEssential(name);
	}
	return super.isFamilyEssential(name);
	}

	@Override
	public byte[] toByteArray() throws IOException {
	return super.toByteArray();
	}

	public Filter getInnerFilter() {
	return userFilter;
	}
	}