You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@omid.apache.org by oh...@apache.org on 2018/08/07 11:59:00 UTC
[1/2] incubator-omid git commit: [OMID-102] Support for user filter
in get/scan operations. snapshot filtering is done by a coprocessor before
user filter is called
Repository: incubator-omid
Updated Branches:
refs/heads/phoenix-integration 1fdd260e5 -> 395fb46f2
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionVisibilityFilter.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionVisibilityFilter.java b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionVisibilityFilter.java
new file mode 100644
index 0000000..20a9846
--- /dev/null
+++ b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionVisibilityFilter.java
@@ -0,0 +1,249 @@
+/*
+ * 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 com.google.common.base.Optional;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.KeyValue;
+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.filter.FilterBase;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+public class TransactionVisibilityFilter extends FilterBase {
+
+ // optional sub-filter to apply to visible cells
+ private final Filter userFilter;
+ private final SnapshotFilterImpl snapshotFilter;
+ private final Map<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 SnapshotFilter getSnapshotFilter() {
+ return snapshotFilter;
+ }
+
+ public TransactionVisibilityFilter(Filter cellFilter,
+ SnapshotFilterImpl snapshotFilter,
+ HBaseTransaction hbaseTransaction) {
+ this.userFilter = cellFilter;
+ this.snapshotFilter = snapshotFilter;
+ commitCache = new HashMap<>();
+ 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> ct = getCommitIfInSnapshot(v, CellUtils.isFamilyDeleteCell(v));
+ if (ct.isPresent()) {
+ commitCache.put(v.getTimestamp(), ct.get());
+ 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), ct.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 && hbaseTransaction.getStartTimestamp() >= cachedCommitTS) {
+ return Optional.of(cachedCommitTS);
+ }
+ 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]));
+ if (commitTS <= hbaseTransaction.getStartTimestamp()) {
+ return Optional.of(commitTS);
+ }
+ }
+ }
+
+ return snapshotFilter.getTSIfInSnapshot(v, hbaseTransaction, commitCache);
+ }
+
+
+ @Override
+ public void reset() throws IOException {
+ commitCache.clear();
+ 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 KeyValue getNextKeyHint(KeyValue currentKV) throws IOException {
+ if (userFilter != null) {
+ return userFilter.getNextKeyHint(currentKV);
+ }
+ return super.getNextKeyHint(currentKV);
+ }
+
+ @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();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-coprocessor/src/test/java/org/apache/omid/transaction/TestSnapshotFilter.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/test/java/org/apache/omid/transaction/TestSnapshotFilter.java b/hbase-coprocessor/src/test/java/org/apache/omid/transaction/TestSnapshotFilter.java
index 29f0a4b..aeae112 100644
--- a/hbase-coprocessor/src/test/java/org/apache/omid/transaction/TestSnapshotFilter.java
+++ b/hbase-coprocessor/src/test/java/org/apache/omid/transaction/TestSnapshotFilter.java
@@ -17,15 +17,18 @@
*/
package org.apache.omid.transaction;
+import com.google.common.util.concurrent.ListenableFuture;
import com.google.inject.Guice;
import com.google.inject.Injector;
import org.apache.hadoop.conf.Configuration;
+
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MiniHBaseCluster;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
@@ -33,6 +36,14 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
+
+import org.apache.hadoop.hbase.filter.BinaryComparator;
+import org.apache.hadoop.hbase.filter.CompareFilter;
+import org.apache.hadoop.hbase.filter.FamilyFilter;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
+import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.omid.TestUtils;
import org.apache.omid.committable.CommitTable;
@@ -41,6 +52,9 @@ import org.apache.omid.metrics.NullMetricsProvider;
import org.apache.omid.timestamp.storage.HBaseTimestampStorageConfig;
import org.apache.omid.tso.TSOServer;
import org.apache.omid.tso.TSOServerConfig;
+
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterClass;
@@ -49,7 +63,11 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -170,9 +188,9 @@ public class TestSnapshotFilter {
.build();
}
+
@Test(timeOut = 60_000)
public void testGetFirstResult() throws Throwable {
-
byte[] rowName1 = Bytes.toBytes("row1");
byte[] famName1 = Bytes.toBytes(TEST_FAMILY);
byte[] colName1 = Bytes.toBytes("col1");
@@ -225,15 +243,243 @@ public class TestSnapshotFilter {
tt.close();
}
+
+ // This test will fail if filtering is done before snapshot filtering
@Test(timeOut = 60_000)
- public void testGetSecondResult() throws Throwable {
+ public void testServerSideSnapshotFiltering() throws Throwable {
+ byte[] rowName1 = Bytes.toBytes("row1");
+ byte[] famName1 = Bytes.toBytes(TEST_FAMILY);
+ byte[] colName1 = Bytes.toBytes("col1");
+ byte[] dataValue1 = Bytes.toBytes("testWrite-1");
+ byte[] dataValue2 = Bytes.toBytes("testWrite-2");
+
+ String TEST_TABLE = "testServerSideSnapshotFiltering";
+ createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY));
+
+ TTable tt = new TTable(hbaseConf, TEST_TABLE);
+
+ Transaction tx1 = tm.begin();
+ Put put1 = new Put(rowName1);
+ put1.add(famName1, colName1, dataValue1);
+ tt.put(tx1, put1);
+ tm.commit(tx1);
+
+ Transaction tx2 = tm.begin();
+ Put put2 = new Put(rowName1);
+ put2.add(famName1, colName1, dataValue2);
+ tt.put(tx2, put2);
+
+ Transaction tx3 = tm.begin();
+ Get get = new Get(rowName1);
+
+ // If snapshot filtering is not done in the server then the first value is
+ // "testWrite-2" and the whole row will be filtered out.
+ SingleColumnValueFilter filter = new SingleColumnValueFilter(
+ famName1,
+ colName1,
+ CompareFilter.CompareOp.EQUAL,
+ new SubstringComparator("testWrite-1"));
+
+ get.setFilter(filter);
+ Result results = tt.get(tx3, get);
+ assertTrue(results.size() == 1);
+ }
+
+ @Test(timeOut = 60_000)
+ public void testGetWithFamilyDelete() throws Throwable {
byte[] rowName1 = Bytes.toBytes("row1");
byte[] famName1 = Bytes.toBytes(TEST_FAMILY);
+ byte[] famName2 = Bytes.toBytes("test-fam2");
byte[] colName1 = Bytes.toBytes("col1");
+ byte[] colName2 = Bytes.toBytes("col2");
byte[] dataValue1 = Bytes.toBytes("testWrite-1");
- String TEST_TABLE = "testGetFirstResult";
+ String TEST_TABLE = "testGetWithFamilyDelete";
+ createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY), famName2);
+ TTable tt = new TTable(hbaseConf, TEST_TABLE);
+
+ Transaction tx1 = tm.begin();
+
+ Put put1 = new Put(rowName1);
+ put1.add(famName1, colName1, dataValue1);
+ tt.put(tx1, put1);
+
+ tm.commit(tx1);
+
+ Transaction tx2 = tm.begin();
+ Put put2 = new Put(rowName1);
+ put2.add(famName2, colName2, dataValue1);
+ tt.put(tx2, put2);
+ tm.commit(tx2);
+
+ Transaction tx3 = tm.begin();
+
+ Delete d = new Delete(rowName1);
+ d.deleteFamily(famName2);
+ tt.delete(tx3, d);
+
+
+ Transaction tx4 = tm.begin();
+
+ Get get = new Get(rowName1);
+
+ Filter filter1 = new FilterList(FilterList.Operator.MUST_PASS_ONE,
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(TEST_FAMILY))),
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(famName2)));
+
+ get.setFilter(filter1);
+ Result result = tt.get(tx4, get);
+ assertTrue(result.size() == 2, "Result should be 2");
+
+ tm.commit(tx3);
+
+ Transaction tx5 = tm.begin();
+ result = tt.get(tx5, get);
+ assertTrue(result.size() == 1, "Result should be 1");
+
+ tt.close();
+ }
+
+ @Test(timeOut = 60_000)
+ public void testReadFromCommitTable() throws Exception {
+ final byte[] rowName1 = Bytes.toBytes("row1");
+ byte[] famName1 = Bytes.toBytes(TEST_FAMILY);
+ byte[] colName1 = Bytes.toBytes("col1");
+ byte[] dataValue1 = Bytes.toBytes("testWrite-1");
+ final String TEST_TABLE = "testReadFromCommitTable";
+ final byte[] famName2 = Bytes.toBytes("test-fam2");
+
+ final CountDownLatch readAfterCommit = new CountDownLatch(1);
+ final CountDownLatch postCommitBegin = new CountDownLatch(1);
+
+ final AtomicBoolean readFailed = new AtomicBoolean(false);
+ final AbstractTransactionManager tm = (AbstractTransactionManager) newTransactionManager();
+ createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY), famName2);
+
+ doAnswer(new Answer<ListenableFuture<Void>>() {
+ @Override
+ public ListenableFuture<Void> answer(InvocationOnMock invocation) throws Throwable {
+ LOG.info("Releasing readAfterCommit barrier");
+ readAfterCommit.countDown();
+ LOG.info("Waiting postCommitBegin barrier");
+ postCommitBegin.await();
+ ListenableFuture<Void> result = (ListenableFuture<Void>) invocation.callRealMethod();
+ return result;
+ }
+ }).when(syncPostCommitter).updateShadowCells(any(HBaseTransaction.class));
+
+ Thread readThread = new Thread("Read Thread") {
+ @Override
+ public void run() {
+
+ try {
+ LOG.info("Waiting readAfterCommit barrier");
+ readAfterCommit.await();
+
+ Transaction tx4 = tm.begin();
+ TTable tt = new TTable(hbaseConf, TEST_TABLE);
+ Get get = new Get(rowName1);
+
+ Filter filter1 = new FilterList(FilterList.Operator.MUST_PASS_ONE,
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(TEST_FAMILY))),
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(famName2)));
+
+ get.setFilter(filter1);
+ Result result = tt.get(tx4, get);
+
+ if (result.size() == 2) {
+ readFailed.set(false);
+ }
+ else {
+ readFailed.set(false);
+ }
+
+ postCommitBegin.countDown();
+ } catch (Throwable e) {
+ readFailed.set(false);
+ LOG.error("Error whilst reading", e);
+ }
+ }
+ };
+ readThread.start();
+
+ TTable table = new TTable(hbaseConf, TEST_TABLE);
+ final HBaseTransaction t1 = (HBaseTransaction) tm.begin();
+ Put put1 = new Put(rowName1);
+ put1.add(famName1, colName1, dataValue1);
+ table.put(t1, put1);
+ tm.commit(t1);
+
+ readThread.join();
+
+ assertFalse(readFailed.get(), "Read should have succeeded");
+
+ }
+
+
+
+ @Test(timeOut = 60_000)
+ public void testGetWithFilter() throws Throwable {
+ byte[] rowName1 = Bytes.toBytes("row1");
+ byte[] famName1 = Bytes.toBytes(TEST_FAMILY);
+ byte[] famName2 = Bytes.toBytes("test-fam2");
+ byte[] colName1 = Bytes.toBytes("col1");
+ byte[] colName2 = Bytes.toBytes("col2");
+ byte[] dataValue1 = Bytes.toBytes("testWrite-1");
+
+ String TEST_TABLE = "testGetWithFilter";
+ createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY), famName2);
+ TTable tt = new TTable(hbaseConf, TEST_TABLE);
+
+ Transaction tx1 = tm.begin();
+
+ Put put1 = new Put(rowName1);
+ put1.add(famName1, colName1, dataValue1);
+ tt.put(tx1, put1);
+
+ tm.commit(tx1);
+
+ Transaction tx2 = tm.begin();
+ Put put2 = new Put(rowName1);
+ put2.add(famName2, colName2, dataValue1);
+ tt.put(tx2, put2);
+ tm.commit(tx2);
+
+ Transaction tx3 = tm.begin();
+
+ Get get = new Get(rowName1);
+
+ Filter filter1 = new FilterList(FilterList.Operator.MUST_PASS_ONE,
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(TEST_FAMILY))),
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(famName2)));
+
+ get.setFilter(filter1);
+ Result result = tt.get(tx3, get);
+ assertTrue(result.size() == 2, "Result should be 2");
+
+
+ Filter filter2 = new FilterList(FilterList.Operator.MUST_PASS_ONE,
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(TEST_FAMILY))));
+
+ get.setFilter(filter2);
+ result = tt.get(tx3, get);
+ assertTrue(result.size() == 1, "Result should be 2");
+
+ tm.commit(tx3);
+
+ tt.close();
+ }
+
+
+ @Test(timeOut = 60_000)
+ public void testGetSecondResult() throws Throwable {
+ byte[] rowName1 = Bytes.toBytes("row1");
+ byte[] famName1 = Bytes.toBytes(TEST_FAMILY);
+ byte[] colName1 = Bytes.toBytes("col1");
+ byte[] dataValue1 = Bytes.toBytes("testWrite-1");
+
+ String TEST_TABLE = "testGetSecondResult";
createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY));
TTable tt = new TTable(hbaseConf, TEST_TABLE);
@@ -273,7 +519,7 @@ public class TestSnapshotFilter {
byte[] colName1 = Bytes.toBytes("col1");
byte[] dataValue1 = Bytes.toBytes("testWrite-1");
- String TEST_TABLE = "testGetFirstResult";
+ String TEST_TABLE = "testScanFirstResult";
createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY));
TTable tt = new TTable(hbaseConf, TEST_TABLE);
@@ -318,6 +564,59 @@ public class TestSnapshotFilter {
tt.close();
}
+
+ @Test(timeOut = 60_000)
+ public void testScanWithFilter() throws Throwable {
+
+ byte[] rowName1 = Bytes.toBytes("row1");
+ byte[] famName1 = Bytes.toBytes(TEST_FAMILY);
+ byte[] famName2 = Bytes.toBytes("test-fam2");
+ byte[] colName1 = Bytes.toBytes("col1");
+ byte[] colName2 = Bytes.toBytes("col2");
+ byte[] dataValue1 = Bytes.toBytes("testWrite-1");
+
+ String TEST_TABLE = "testScanWithFilter";
+ createTableIfNotExists(TEST_TABLE, famName1, famName2);
+ TTable tt = new TTable(hbaseConf, TEST_TABLE);
+
+ Transaction tx1 = tm.begin();
+ Put put1 = new Put(rowName1);
+ put1.add(famName1, colName1, dataValue1);
+ tt.put(tx1, put1);
+ tm.commit(tx1);
+
+ Transaction tx2 = tm.begin();
+ Put put2 = new Put(rowName1);
+ put2.add(famName2, colName2, dataValue1);
+ tt.put(tx2, put2);
+
+ tm.commit(tx2);
+ Transaction tx3 = tm.begin();
+
+ Scan scan = new Scan();
+ scan.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ONE,
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(TEST_FAMILY)))));
+ scan.setStartRow(rowName1).setStopRow(rowName1);
+
+ ResultScanner iterableRS = tt.getScanner(tx3, scan);
+ Result result = iterableRS.next();
+ assertTrue(result.containsColumn(famName1, colName1));
+ assertFalse(result.containsColumn(famName2, colName2));
+
+ scan.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ONE,
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(TEST_FAMILY))),
+ new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(famName2))));
+
+ iterableRS = tt.getScanner(tx3, scan);
+ result = iterableRS.next();
+ assertTrue(result.containsColumn(famName1, colName1));
+ assertTrue(result.containsColumn(famName2, colName2));
+
+ tm.commit(tx3);
+ tt.close();
+ }
+
+
@Test(timeOut = 60_000)
public void testScanSecondResult() throws Throwable {
@@ -326,7 +625,7 @@ public class TestSnapshotFilter {
byte[] colName1 = Bytes.toBytes("col1");
byte[] dataValue1 = Bytes.toBytes("testWrite-1");
- String TEST_TABLE = "testGetFirstResult";
+ String TEST_TABLE = "testScanSecondResult";
createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY));
TTable tt = new TTable(hbaseConf, TEST_TABLE);
@@ -370,7 +669,7 @@ public class TestSnapshotFilter {
byte[] dataValue1 = Bytes.toBytes("testWrite-1");
byte[] dataValue2 = Bytes.toBytes("testWrite-2");
- String TEST_TABLE = "testGetFirstResult";
+ String TEST_TABLE = "testScanFewResults";
createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY));
TTable tt = new TTable(hbaseConf, TEST_TABLE);
@@ -420,7 +719,7 @@ public class TestSnapshotFilter {
byte[] dataValue1 = Bytes.toBytes("testWrite-1");
byte[] dataValue2 = Bytes.toBytes("testWrite-2");
- String TEST_TABLE = "testGetFirstResult";
+ String TEST_TABLE = "testScanFewResultsDifferentTransaction";
createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY));
TTable tt = new TTable(hbaseConf, TEST_TABLE);
@@ -473,7 +772,7 @@ public class TestSnapshotFilter {
byte[] dataValue1 = Bytes.toBytes("testWrite-1");
byte[] dataValue2 = Bytes.toBytes("testWrite-2");
- String TEST_TABLE = "testGetFirstResult";
+ String TEST_TABLE = "testScanFewResultsSameTransaction";
createTableIfNotExists(TEST_TABLE, Bytes.toBytes(TEST_FAMILY));
TTable tt = new TTable(hbaseConf, TEST_TABLE);
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
----------------------------------------------------------------------
diff --git a/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java b/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
index a6d63c7..a704453 100644
--- a/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
+++ b/tso-server/src/main/java/org/apache/omid/tso/PersistenceProcessorHandler.java
@@ -30,7 +30,8 @@ import java.io.IOException;
import java.util.concurrent.ExecutionException;
import static com.codahale.metrics.MetricRegistry.name;
-import static org.apache.omid.tso.PersistEvent.Type.*;
+import static org.apache.omid.tso.PersistEvent.Type.COMMIT_RETRY;
+
public class PersistenceProcessorHandler implements WorkHandler<PersistenceProcessorImpl.PersistBatchEvent> {
[2/2] incubator-omid git commit: [OMID-102] Support for user filter
in get/scan operations. snapshot filtering is done by a coprocessor before
user filter is called
Posted by oh...@apache.org.
[OMID-102] Support for user filter in get/scan operations. snapshot filtering is done by a coprocessor before user filter is called
Signed-off-by: Ohad Shacham <oh...@yahoo-inc.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-omid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-omid/commit/395fb46f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-omid/tree/395fb46f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-omid/diff/395fb46f
Branch: refs/heads/phoenix-integration
Commit: 395fb46f2444f20a1addbb07306047ed4edfa454
Parents: 1fdd260
Author: Yonatan Gottesman <yo...@gmail.com>
Authored: Tue Aug 7 14:24:27 2018 +0300
Committer: Ohad Shacham <oh...@yahoo-inc.com>
Committed: Tue Aug 7 14:55:56 2018 +0300
----------------------------------------------------------------------
.../transaction/AttributeSetSnapshotFilter.java | 4 +-
.../transaction/HBaseSyncPostCommitter.java | 2 +-
.../transaction/HBaseTransactionManager.java | 5 +-
.../omid/transaction/HTableAccessWrapper.java | 15 +-
.../apache/omid/transaction/SnapshotFilter.java | 4 +-
.../omid/transaction/SnapshotFilterImpl.java | 220 ++++++++++---
.../org/apache/omid/transaction/TTable.java | 147 ++-------
.../omid/transaction/TableAccessWrapper.java | 12 +-
.../apache/omid/transaction/TestCellUtils.java | 19 +-
.../omid/transaction/TestColumnIterator.java | 6 +-
.../omid/transaction/TestShadowCells.java | 23 +-
.../org/apache/omid/transaction/CellUtils.java | 103 ++++--
.../hbase/regionserver/CompactorScanner.java | 2 +-
.../hbase/regionserver/OmidRegionScanner.java | 78 ++---
.../hbase/regionserver/RegionAccessWrapper.java | 9 +-
.../omid/transaction/OmidSnapshotFilter.java | 135 ++++----
.../omid/transaction/TransactionFilters.java | 30 ++
.../TransactionVisibilityFilter.java | 249 +++++++++++++++
.../omid/transaction/TestSnapshotFilter.java | 315 ++++++++++++++++++-
.../omid/tso/PersistenceProcessorHandler.java | 3 +-
20 files changed, 1011 insertions(+), 370 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/AttributeSetSnapshotFilter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/AttributeSetSnapshotFilter.java b/hbase-client/src/main/java/org/apache/omid/transaction/AttributeSetSnapshotFilter.java
index 482dc4c..f4b6191 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/AttributeSetSnapshotFilter.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/AttributeSetSnapshotFilter.java
@@ -49,7 +49,7 @@ public class AttributeSetSnapshotFilter implements SnapshotFilter {
}
@Override
- public Result get(TTable ttable, Get get, HBaseTransaction transaction) throws IOException {
+ public Result get(Get get, HBaseTransaction transaction) throws IOException {
get.setAttribute(CellUtils.TRANSACTION_ATTRIBUTE, getBuilder(transaction).build().toByteArray());
get.setAttribute(CellUtils.CLIENT_GET_ATTRIBUTE, Bytes.toBytes(true));
@@ -57,7 +57,7 @@ public class AttributeSetSnapshotFilter implements SnapshotFilter {
}
@Override
- public ResultScanner getScanner(TTable ttable, Scan scan, HBaseTransaction transaction) throws IOException {
+ public ResultScanner getScanner(Scan scan, HBaseTransaction transaction) throws IOException {
scan.setAttribute(CellUtils.TRANSACTION_ATTRIBUTE, getBuilder(transaction).build().toByteArray());
return table.getScanner(scan);
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java
index 8049b3a..4b3560f 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseSyncPostCommitter.java
@@ -54,7 +54,7 @@ public class HBaseSyncPostCommitter implements PostCommitActions {
private void addShadowCell(HBaseCellId cell, HBaseTransaction tx, SettableFuture<Void> updateSCFuture) {
Put put = new Put(cell.getRow());
put.add(cell.getFamily(),
- CellUtils.addShadowCellSuffix(cell.getQualifier(), 0, cell.getQualifier().length),
+ CellUtils.addShadowCellSuffixPrefix(cell.getQualifier(), 0, cell.getQualifier().length),
cell.getTimestamp(),
Bytes.toBytes(tx.getCommitTimestamp()));
try {
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java
index 1242eeb..e20f873 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionManager.java
@@ -17,15 +17,12 @@
*/
package org.apache.omid.transaction;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
-import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.omid.committable.CommitTable;
-import org.apache.omid.committable.CommitTable.CommitTimestamp;
import org.apache.omid.committable.hbase.HBaseCommitTable;
import org.apache.omid.committable.hbase.HBaseCommitTableConfig;
import org.apache.omid.tools.hbase.HBaseLogin;
@@ -267,7 +264,7 @@ public class HBaseTransactionManager extends AbstractTransactionManager implemen
Get get = new Get(hBaseCellId.getRow());
byte[] family = hBaseCellId.getFamily();
- byte[] shadowCellQualifier = CellUtils.addShadowCellSuffix(hBaseCellId.getQualifier());
+ byte[] shadowCellQualifier = CellUtils.addShadowCellSuffixPrefix(hBaseCellId.getQualifier());
get.addColumn(family, shadowCellQualifier);
get.setMaxVersions(1);
get.setTimeStamp(startTimestamp);
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/HTableAccessWrapper.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HTableAccessWrapper.java b/hbase-client/src/main/java/org/apache/omid/transaction/HTableAccessWrapper.java
index 1994cad..84c8d2c 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HTableAccessWrapper.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HTableAccessWrapper.java
@@ -17,13 +17,17 @@
*/
package org.apache.omid.transaction;
-import java.io.IOException;
-import java.util.List;
-
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+
+import java.io.IOException;
+import java.util.List;
+
+
// This class wraps the HTableInterface object when doing client side filtering.
public class HTableAccessWrapper implements TableAccessWrapper {
@@ -51,4 +55,9 @@ public class HTableAccessWrapper implements TableAccessWrapper {
writeTable.put(put);
}
+ @Override
+ public ResultScanner getScanner(Scan scan) throws IOException {
+ return readTable.getScanner(scan);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilter.java b/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilter.java
index 4668c9e..4d2b8da 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilter.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilter.java
@@ -32,9 +32,9 @@ import com.google.common.base.Optional;
public interface SnapshotFilter {
- public Result get(TTable ttable, Get get, HBaseTransaction transaction) throws IOException;
+ public Result get(Get get, HBaseTransaction transaction) throws IOException;
- public ResultScanner getScanner(TTable ttable, Scan scan, HBaseTransaction transaction) throws IOException;
+ public ResultScanner getScanner(Scan scan, HBaseTransaction transaction) throws IOException;
public List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
int versionsToRequest, Map<String, Long> familyDeletionCache, Map<String,byte[]> attributeMap) throws IOException;
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilterImpl.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilterImpl.java b/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilterImpl.java
index d0472d2..a048a0b 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilterImpl.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilterImpl.java
@@ -23,21 +23,20 @@ import static org.apache.omid.committable.CommitTable.CommitTimestamp.Location.N
import static org.apache.omid.committable.CommitTable.CommitTimestamp.Location.SHADOW_CELL;
import java.io.IOException;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
@@ -45,10 +44,8 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.omid.committable.CommitTable;
import org.apache.omid.committable.CommitTable.CommitTimestamp;
-import org.apache.omid.proto.TSOProto;
import org.apache.omid.transaction.AbstractTransaction.VisibilityLevel;
import org.apache.omid.transaction.HBaseTransactionManager.CommitTimestampLocatorImpl;
-import org.apache.omid.transaction.TTable.TransactionalClientScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,8 +63,16 @@ public class SnapshotFilterImpl implements SnapshotFilter {
private TableAccessWrapper tableAccessWrapper;
+ public CommitTable.Client getCommitTableClient() {
+ return commitTableClient;
+ }
+
private CommitTable.Client commitTableClient;
-
+
+ public TableAccessWrapper getTableAccessWrapper() {
+ return tableAccessWrapper;
+ }
+
public SnapshotFilterImpl(TableAccessWrapper tableAccessWrapper, CommitTable.Client commitTableClient) throws IOException {
this.tableAccessWrapper = tableAccessWrapper;
this.commitTableClient = commitTableClient;
@@ -108,14 +113,13 @@ public class SnapshotFilterImpl implements SnapshotFilter {
if (familyDeletionCommitTimestamp != null && familyDeletionCommitTimestamp >= cell.getTimestamp()) {
return true;
}
-
return false;
}
private void healShadowCell(Cell cell, long commitTimestamp) {
Put put = new Put(CellUtil.cloneRow(cell));
byte[] family = CellUtil.cloneFamily(cell);
- byte[] shadowCellQualifier = CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
+ byte[] shadowCellQualifier = CellUtils.addShadowCellSuffixPrefix(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength());
put.add(family, shadowCellQualifier, cell.getTimestamp(), Bytes.toBytes(commitTimestamp));
@@ -220,7 +224,7 @@ public class SnapshotFilterImpl implements SnapshotFilter {
}
- private Optional<Long> tryToLocateCellCommitTimestamp(long epoch,
+ public Optional<Long> tryToLocateCellCommitTimestamp(long epoch,
Cell cell,
Map<Long, Long> commitCache)
throws IOException {
@@ -295,6 +299,7 @@ public class SnapshotFilterImpl implements SnapshotFilter {
return commitCache;
}
+
private void buildFamilyDeletionCache(HBaseTransaction transaction, List<Cell> rawCells, Map<String, Long> familyDeletionCache, Map<Long, Long> commitCache, Map<String,byte[]> attributeMap) throws IOException {
for (Cell cell : rawCells) {
if (CellUtils.isFamilyDeleteCell(cell)) {
@@ -303,7 +308,7 @@ public class SnapshotFilterImpl implements SnapshotFilter {
if (familyDeletionCache.containsKey(key))
return;
- Optional<Long> commitTimeStamp = getTSIfInTransaction(cell, transaction, commitCache);
+ Optional<Long> commitTimeStamp = getTSIfInTransaction(cell, transaction);
if (!commitTimeStamp.isPresent()) {
commitTimeStamp = getTSIfInSnapshot(cell, transaction, commitCache);
@@ -343,8 +348,8 @@ public class SnapshotFilterImpl implements SnapshotFilter {
}
}
- private Optional<Long> getTSIfInTransaction(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache) {
+ public Optional<Long> getTSIfInTransaction(Cell kv, HBaseTransaction transaction) {
long startTimestamp = transaction.getStartTimestamp();
long readTimestamp = transaction.getReadTimestamp();
@@ -358,7 +363,8 @@ public class SnapshotFilterImpl implements SnapshotFilter {
return Optional.absent();
}
- private Optional<Long> getTSIfInSnapshot(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache)
+
+ public Optional<Long> getTSIfInSnapshot(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache)
throws IOException {
Optional<Long> commitTimestamp = getCommitTimestamp(kv, transaction, commitCache);
@@ -373,7 +379,7 @@ public class SnapshotFilterImpl implements SnapshotFilter {
Get pendingGet = new Get(CellUtil.cloneRow(cell));
pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell));
- pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
+ pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtils.addShadowCellSuffixPrefix(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength()));
pendingGet.setMaxVersions(versionCount);
@@ -410,36 +416,40 @@ public class SnapshotFilterImpl implements SnapshotFilter {
Map<Long, Long> commitCache = buildCommitCache(rawCells);
buildFamilyDeletionCache(transaction, rawCells, familyDeletionCache, commitCache, attributeMap);
- for (Collection<Cell> columnCells : groupCellsByColumnFilteringShadowCellsAndFamilyDeletion(rawCells)) {
+ ImmutableList<Collection<Cell>> filteredCells;
+ if (transaction.getVisibilityLevel() == VisibilityLevel.SNAPSHOT_ALL) {
+ filteredCells = groupCellsByColumnFilteringShadowCells(rawCells);
+ } else {
+ filteredCells = groupCellsByColumnFilteringShadowCellsAndFamilyDeletion(rawCells);
+ }
+
+ for (Collection<Cell> columnCells : filteredCells) {
boolean snapshotValueFound = false;
Cell oldestCell = null;
for (Cell cell : columnCells) {
- snapshotValueFound = checkFamilyDeletionCache(cell, transaction, familyDeletionCache, commitCache);
+ oldestCell = cell;
+ if (getTSIfInTransaction(cell, transaction).isPresent() ||
+ getTSIfInSnapshot(cell, transaction, commitCache).isPresent()) {
- if (snapshotValueFound == true) {
if (transaction.getVisibilityLevel() == VisibilityLevel.SNAPSHOT_ALL) {
- snapshotValueFound = false;
- } else {
- break;
- }
- }
-
- if (getTSIfInTransaction(cell, transaction, commitCache).isPresent() ||
- getTSIfInSnapshot(cell, transaction, commitCache).isPresent()) {
- if (!CellUtil.matchingValue(cell, CellUtils.DELETE_TOMBSTONE)) {
keyValuesInSnapshot.add(cell);
- }
-
- // We can finish looking for additional results in two cases:
- // 1. if we found a result and we are not in SNAPSHOT_ALL mode.
- // 2. if we found a result that was not written by the current transaction.
- if (transaction.getVisibilityLevel() != VisibilityLevel.SNAPSHOT_ALL ||
- !getTSIfInTransaction(cell, transaction, commitCache).isPresent()) {
+ if (getTSIfInTransaction(cell, transaction).isPresent()) {
+ snapshotValueFound = false;
+ continue;
+ } else {
+ snapshotValueFound = true;
+ break;
+ }
+ } else {
+ if (!checkFamilyDeletionCache(cell, transaction, familyDeletionCache, commitCache) &&
+ !CellUtils.isTombstone(cell)) {
+ keyValuesInSnapshot.add(cell);
+ }
snapshotValueFound = true;
break;
+
}
}
- oldestCell = cell;
}
if (!snapshotValueFound) {
assert (oldestCell != null);
@@ -467,21 +477,21 @@ public class SnapshotFilterImpl implements SnapshotFilter {
}
@Override
- public Result get(TTable ttable, Get get, HBaseTransaction transaction) throws IOException {
+ public Result get(Get get, HBaseTransaction transaction) throws IOException {
Result result = tableAccessWrapper.get(get);
List<Cell> filteredKeyValues = Collections.emptyList();
if (!result.isEmpty()) {
- filteredKeyValues = ttable.filterCellsForSnapshot(result.listCells(), transaction, get.getMaxVersions(), new HashMap<String, Long>(), get.getAttributesMap());
+ filteredKeyValues = filterCellsForSnapshot(result.listCells(), transaction, get.getMaxVersions(), new HashMap<String, Long>(), get.getAttributesMap());
}
return Result.create(filteredKeyValues);
}
@Override
- public ResultScanner getScanner(TTable ttable, Scan scan, HBaseTransaction transaction) throws IOException {
+ public ResultScanner getScanner(Scan scan, HBaseTransaction transaction) throws IOException {
- return ttable.new TransactionalClientScanner(transaction, scan, 1);
+ return new TransactionalClientScanner(transaction, scan, 1);
}
@@ -519,8 +529,7 @@ public class SnapshotFilterImpl implements SnapshotFilter {
@Override
public boolean apply(Cell cell) {
- boolean familyDeletionMarkerCondition = CellUtil.matchingQualifier(cell, CellUtils.FAMILY_DELETE_QUALIFIER) &&
- CellUtil.matchingValue(cell, HConstants.EMPTY_BYTE_ARRAY);
+ boolean familyDeletionMarkerCondition = CellUtils.isFamilyDeleteCell(cell);
return cell != null && !CellUtils.isShadowCell(cell) && !familyDeletionMarkerCondition;
}
@@ -541,4 +550,137 @@ public class SnapshotFilterImpl implements SnapshotFilter {
.asList();
}
+
+ static ImmutableList<Collection<Cell>> groupCellsByColumnFilteringShadowCells(List<Cell> rawCells) {
+
+ Predicate<Cell> shadowCellFilter = new Predicate<Cell>() {
+ @Override
+ public boolean apply(Cell cell) {
+ return cell != null && !CellUtils.isShadowCell(cell);
+ }
+ };
+
+ Function<Cell, ColumnWrapper> cellToColumnWrapper = new Function<Cell, ColumnWrapper>() {
+
+ @Override
+ public ColumnWrapper apply(Cell cell) {
+ return new ColumnWrapper(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell));
+ }
+
+ };
+
+ return Multimaps.index(Iterables.filter(rawCells, shadowCellFilter), cellToColumnWrapper)
+ .asMap().values()
+ .asList();
+ }
+
+
+ public class TransactionalClientScanner implements ResultScanner {
+
+ private HBaseTransaction state;
+ private ResultScanner innerScanner;
+ private int maxVersions;
+ Map<String, Long> familyDeletionCache;
+ private Map<String,byte[]> attributeMap;
+
+ TransactionalClientScanner(HBaseTransaction state, Scan scan, int maxVersions)
+ throws IOException {
+ if (scan.hasFilter()) {
+ LOG.warn("Client scanner with filter will return un expected results. Use Coprocessor scanning");
+ }
+ this.state = state;
+ this.innerScanner = tableAccessWrapper.getScanner(scan);
+ this.maxVersions = maxVersions;
+ this.familyDeletionCache = new HashMap<String, Long>();
+ this.attributeMap = scan.getAttributesMap();
+ }
+
+
+ @Override
+ public Result next() throws IOException {
+ List<Cell> filteredResult = Collections.emptyList();
+ while (filteredResult.isEmpty()) {
+ Result result = innerScanner.next();
+ if (result == null) {
+ return null;
+ }
+ if (!result.isEmpty()) {
+ filteredResult = filterCellsForSnapshot(result.listCells(), state, maxVersions, familyDeletionCache, attributeMap);
+ }
+ }
+ return Result.create(filteredResult);
+ }
+
+ // In principle no need to override, copied from super.next(int) to make
+ // sure it works even if super.next(int)
+ // changes its implementation
+ @Override
+ public Result[] next(int nbRows) throws IOException {
+ // Collect values to be returned here
+ ArrayList<Result> resultSets = new ArrayList<>(nbRows);
+ for (int i = 0; i < nbRows; i++) {
+ Result next = next();
+ if (next != null) {
+ resultSets.add(next);
+ } else {
+ break;
+ }
+ }
+ return resultSets.toArray(new Result[resultSets.size()]);
+ }
+
+ @Override
+ public void close() {
+ innerScanner.close();
+ }
+
+ @Override
+ public Iterator<Result> iterator() {
+ return new ResultIterator(this);
+ }
+
+ // ------------------------------------------------------------------------------------------------------------
+ // --------------------------------- Helper class for TransactionalClientScanner ------------------------------
+ // ------------------------------------------------------------------------------------------------------------
+
+ class ResultIterator implements Iterator<Result> {
+
+ TransactionalClientScanner scanner;
+ Result currentResult;
+
+ ResultIterator(TransactionalClientScanner scanner) {
+ try {
+ this.scanner = scanner;
+ currentResult = scanner.next();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return currentResult != null && !currentResult.isEmpty();
+ }
+
+ @Override
+ public Result next() {
+ try {
+ Result result = currentResult;
+ currentResult = scanner.next();
+ return result;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void remove() {
+ throw new RuntimeException("Not implemented");
+ }
+
+ }
+
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/TTable.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/TTable.java b/hbase-client/src/main/java/org/apache/omid/transaction/TTable.java
index 1331b87..4813d5b 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/TTable.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/TTable.java
@@ -19,10 +19,6 @@ package org.apache.omid.transaction;
import java.io.Closeable;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -107,14 +103,22 @@ public class TTable implements Closeable {
table = hTable;
healerTable = new HTable(table.getConfiguration(), table.getTableName());
this.serverSideFilter = serverSideFilter;
- snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) : new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable));
+ snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) :
+ new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable));
+ }
+
+ public TTable(HTableInterface hTable, SnapshotFilter snapshotFilter ) throws IOException {
+ table = hTable;
+ healerTable = new HTable(table.getConfiguration(), table.getTableName());
+ this.snapshotFilter = snapshotFilter;
}
public TTable(HTableInterface hTable, CommitTable.Client commitTableClient) throws IOException {
table = hTable;
healerTable = new HTable(table.getConfiguration(), table.getTableName());
serverSideFilter = table.getConfiguration().getBoolean("omid.server.side.filter", false);
- snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) : new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable), commitTableClient);
+ snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) :
+ new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable), commitTableClient);
}
public TTable(HTableInterface hTable, HTableInterface healerTable) throws IOException {
@@ -122,14 +126,16 @@ public class TTable implements Closeable {
this.healerTable = healerTable;
Configuration config = table.getConfiguration();
serverSideFilter = (config == null) ? false : config.getBoolean("omid.server.side.filter", false);
- snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) : new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable));
+ snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) :
+ new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable));
}
public TTable(HTableInterface hTable, HTableInterface healerTable, CommitTable.Client commitTableClient) throws IOException {
table = hTable;
this.healerTable = healerTable;
serverSideFilter = table.getConfiguration().getBoolean("omid.server.side.filter", false);
- snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) : new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable), commitTableClient);
+ snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) :
+ new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable), commitTableClient);
}
@@ -182,15 +188,15 @@ public class TTable implements Closeable {
} else {
for (byte[] qualifier : qualifiers) {
tsget.addColumn(family, qualifier);
- tsget.addColumn(family, CellUtils.addShadowCellSuffix(qualifier));
+ tsget.addColumn(family, CellUtils.addShadowCellSuffixPrefix(qualifier));
}
tsget.addColumn(family, CellUtils.FAMILY_DELETE_QUALIFIER);
- tsget.addColumn(family, CellUtils.addShadowCellSuffix(CellUtils.FAMILY_DELETE_QUALIFIER));
+ tsget.addColumn(family, CellUtils.addShadowCellSuffixPrefix(CellUtils.FAMILY_DELETE_QUALIFIER));
}
}
LOG.trace("Initial Get = {}", tsget);
- return snapshotFilter.get(this, tsget, transaction);
+ return snapshotFilter.get(tsget, transaction);
}
static private void propagateAttributes(OperationWithAttributes from, OperationWithAttributes to) {
@@ -346,7 +352,7 @@ public class TTable implements Closeable {
Bytes.putLong(kv.getValueArray(), kv.getTimestampOffset(), timestamp);
tsput.add(kv);
tsput.add(CellUtil.cloneFamily(kv),
- CellUtils.addShadowCellSuffix(CellUtil.cloneQualifier(kv), 0, CellUtil.cloneQualifier(kv).length),
+ CellUtils.addShadowCellSuffixPrefix(CellUtil.cloneQualifier(kv), 0, CellUtil.cloneQualifier(kv).length),
kv.getTimestamp(),
Bytes.toBytes(commitTimestamp));
}
@@ -386,7 +392,7 @@ public class TTable implements Closeable {
if (autoCommit) {
tsput.add(CellUtil.cloneFamily(kv),
- CellUtils.addShadowCellSuffix(CellUtil.cloneQualifier(kv), 0, CellUtil.cloneQualifier(kv).length),
+ CellUtils.addShadowCellSuffixPrefix(CellUtil.cloneQualifier(kv), 0, CellUtil.cloneQualifier(kv).length),
kv.getTimestamp(),
Bytes.toBytes(kv.getTimestamp()));
} else {
@@ -435,125 +441,14 @@ public class TTable implements Closeable {
continue;
}
for (byte[] qualifier : qualifiers) {
- tsscan.addColumn(family, CellUtils.addShadowCellSuffix(qualifier));
+ tsscan.addColumn(family, CellUtils.addShadowCellSuffixPrefix(qualifier));
}
if (!qualifiers.isEmpty()) {
tsscan.addColumn(entry.getKey(), CellUtils.FAMILY_DELETE_QUALIFIER);
}
}
- return snapshotFilter.getScanner(this, tsscan, transaction);
- }
-
-
- List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
- int versionsToRequest, Map<String, Long> familyDeletionCache, Map<String,byte[]> attributeMap) throws IOException {
- return snapshotFilter.filterCellsForSnapshot(rawCells, transaction, versionsToRequest, familyDeletionCache, attributeMap);
- }
-
-
- public class TransactionalClientScanner implements ResultScanner {
-
- private HBaseTransaction state;
- private ResultScanner innerScanner;
- private int maxVersions;
- Map<String, Long> familyDeletionCache;
- private Map<String,byte[]> attributeMap;
-
- TransactionalClientScanner(HBaseTransaction state, Scan scan, int maxVersions)
- throws IOException {
- this.state = state;
- this.innerScanner = table.getScanner(scan);
- this.maxVersions = maxVersions;
- this.familyDeletionCache = new HashMap<String, Long>();
- this.attributeMap = scan.getAttributesMap();
- }
-
-
- @Override
- public Result next() throws IOException {
- List<Cell> filteredResult = Collections.emptyList();
- while (filteredResult.isEmpty()) {
- Result result = innerScanner.next();
- if (result == null) {
- return null;
- }
- if (!result.isEmpty()) {
- filteredResult = filterCellsForSnapshot(result.listCells(), state, maxVersions, familyDeletionCache, attributeMap);
- }
- }
- return Result.create(filteredResult);
- }
-
- // In principle no need to override, copied from super.next(int) to make
- // sure it works even if super.next(int)
- // changes its implementation
- @Override
- public Result[] next(int nbRows) throws IOException {
- // Collect values to be returned here
- ArrayList<Result> resultSets = new ArrayList<>(nbRows);
- for (int i = 0; i < nbRows; i++) {
- Result next = next();
- if (next != null) {
- resultSets.add(next);
- } else {
- break;
- }
- }
- return resultSets.toArray(new Result[resultSets.size()]);
- }
-
- @Override
- public void close() {
- innerScanner.close();
- }
-
- @Override
- public Iterator<Result> iterator() {
- return new ResultIterator(this);
- }
-
- // ------------------------------------------------------------------------------------------------------------
- // --------------------------------- Helper class for TransactionalClientScanner ------------------------------
- // ------------------------------------------------------------------------------------------------------------
-
- class ResultIterator implements Iterator<Result> {
-
- TransactionalClientScanner scanner;
- Result currentResult;
-
- ResultIterator(TransactionalClientScanner scanner) {
- try {
- this.scanner = scanner;
- currentResult = scanner.next();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public boolean hasNext() {
- return currentResult != null && !currentResult.isEmpty();
- }
-
- @Override
- public Result next() {
- try {
- Result result = currentResult;
- currentResult = scanner.next();
- return result;
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void remove() {
- throw new RuntimeException("Not implemented");
- }
-
- }
-
+ return snapshotFilter.getScanner(tsscan, transaction);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/main/java/org/apache/omid/transaction/TableAccessWrapper.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/TableAccessWrapper.java b/hbase-client/src/main/java/org/apache/omid/transaction/TableAccessWrapper.java
index 2fc5362..050aed0 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/TableAccessWrapper.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/TableAccessWrapper.java
@@ -17,12 +17,16 @@
*/
package org.apache.omid.transaction;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+
import java.io.IOException;
import java.util.List;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Get;
+
//This interface is used to wrap the HTableInterface and Region object when doing client and server side filtering accordingly.
public interface TableAccessWrapper {
@@ -30,5 +34,5 @@ public interface TableAccessWrapper {
public Result[] get(List<Get> get) throws IOException;
public Result get(Get get) throws IOException;
public void put(Put put) throws IOException;
-
+ public ResultScanner getScanner(Scan scan) throws IOException;
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/test/java/org/apache/omid/transaction/TestCellUtils.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/test/java/org/apache/omid/transaction/TestCellUtils.java b/hbase-client/src/test/java/org/apache/omid/transaction/TestCellUtils.java
index 2fdd6a9..8a689ca 100644
--- a/hbase-client/src/test/java/org/apache/omid/transaction/TestCellUtils.java
+++ b/hbase-client/src/test/java/org/apache/omid/transaction/TestCellUtils.java
@@ -31,6 +31,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.SortedMap;
+import static org.apache.omid.transaction.CellUtils.SHADOW_CELL_PREFIX;
import static org.apache.omid.transaction.CellUtils.SHADOW_CELL_SUFFIX;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -99,11 +100,11 @@ public class TestCellUtils {
public void testCorrectMapingOfCellsToShadowCells() throws IOException {
// Create the required data
final byte[] validShadowCellQualifier =
- com.google.common.primitives.Bytes.concat(qualifier, SHADOW_CELL_SUFFIX);
+ com.google.common.primitives.Bytes.concat(SHADOW_CELL_PREFIX, qualifier, SHADOW_CELL_SUFFIX);
final byte[] qualifier2 = Bytes.toBytes("test-qual2");
final byte[] validShadowCellQualifier2 =
- com.google.common.primitives.Bytes.concat(qualifier2, SHADOW_CELL_SUFFIX);
+ com.google.common.primitives.Bytes.concat(SHADOW_CELL_PREFIX, qualifier2, SHADOW_CELL_SUFFIX);
final byte[] qualifier3 = Bytes.toBytes("test-qual3");
@@ -172,10 +173,10 @@ public class TestCellUtils {
public void testShadowCellSuffixConcatenationToQualifier() {
Cell cell = new KeyValue(row, family, qualifier, 1, Bytes.toBytes("value"));
- byte[] suffixedQualifier = CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
+ byte[] suffixedQualifier = CellUtils.addShadowCellSuffixPrefix(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength());
- byte[] expectedQualifier = com.google.common.primitives.Bytes.concat(qualifier, SHADOW_CELL_SUFFIX);
+ byte[] expectedQualifier = com.google.common.primitives.Bytes.concat(SHADOW_CELL_PREFIX, qualifier, SHADOW_CELL_SUFFIX);
assertEquals(suffixedQualifier, expectedQualifier);
}
@@ -184,19 +185,19 @@ public class TestCellUtils {
public void testShadowCellSuffixRemovalFromQualifier(byte[] shadowCellSuffixToTest) throws IOException {
// Test removal from a correclty suffixed qualifier
- byte[] suffixedQualifier = com.google.common.primitives.Bytes.concat(qualifier, shadowCellSuffixToTest);
+ byte[] suffixedQualifier = com.google.common.primitives.Bytes.concat(SHADOW_CELL_PREFIX, qualifier, shadowCellSuffixToTest);
Cell cell = new KeyValue(row, family, suffixedQualifier, 1, Bytes.toBytes("value"));
- byte[] resultedQualifier = CellUtils.removeShadowCellSuffix(cell.getQualifierArray(),
+ byte[] resultedQualifier = CellUtils.removeShadowCellSuffixPrefix(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength());
byte[] expectedQualifier = qualifier;
assertEquals(resultedQualifier, expectedQualifier);
// Test removal from a badly suffixed qualifier
- byte[] badlySuffixedQualifier = com.google.common.primitives.Bytes.concat(qualifier, Bytes.toBytes("BAD"));
+ byte[] badlySuffixedQualifier = com.google.common.primitives.Bytes.concat(SHADOW_CELL_PREFIX, qualifier, Bytes.toBytes("BAD"));
Cell badCell = new KeyValue(row, family, badlySuffixedQualifier, 1, Bytes.toBytes("value"));
try {
- CellUtils.removeShadowCellSuffix(badCell.getQualifierArray(),
+ CellUtils.removeShadowCellSuffixPrefix(badCell.getQualifierArray(),
badCell.getQualifierOffset(),
badCell.getQualifierLength());
fail();
@@ -215,7 +216,7 @@ public class TestCellUtils {
@Test(dataProvider = "shadow-cell-suffixes", timeOut = 10_000)
public void testQualifierLengthFromShadowCellQualifier(byte[] shadowCellSuffixToTest) {
// Test suffixed qualifier
- byte[] suffixedQualifier = com.google.common.primitives.Bytes.concat(qualifier, shadowCellSuffixToTest);
+ byte[] suffixedQualifier = com.google.common.primitives.Bytes.concat(SHADOW_CELL_PREFIX, qualifier, shadowCellSuffixToTest);
int originalQualifierLength =
CellUtils.qualifierLengthFromShadowCellQualifier(suffixedQualifier, 0, suffixedQualifier.length);
assertEquals(originalQualifierLength, qualifier.length);
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/test/java/org/apache/omid/transaction/TestColumnIterator.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/test/java/org/apache/omid/transaction/TestColumnIterator.java b/hbase-client/src/test/java/org/apache/omid/transaction/TestColumnIterator.java
index 5b72856..f7281cd 100644
--- a/hbase-client/src/test/java/org/apache/omid/transaction/TestColumnIterator.java
+++ b/hbase-client/src/test/java/org/apache/omid/transaction/TestColumnIterator.java
@@ -47,13 +47,13 @@ public class TestColumnIterator {
// Group 1 (3 elems but grouping should filter shadow cell, so check for 2)
new KeyValue(row, family1, qualifier1, 0, data),
new KeyValue(row, family1, qualifier1, 1, data),
- new KeyValue(row, family1, CellUtils.addShadowCellSuffix(qualifier1), 0, data),
+ new KeyValue(row, family1, CellUtils.addShadowCellSuffixPrefix(qualifier1), 0, data),
// Group 2 (2 elems but grouping should filter shadow cell, so check for 1)
new KeyValue(row, family1, qualifier2, 0, data),
- new KeyValue(row, family1, CellUtils.addShadowCellSuffix(qualifier2), 0, data),
+ new KeyValue(row, family1, CellUtils.addShadowCellSuffixPrefix(qualifier2), 0, data),
// Group 3 (2 elems but grouping should filter shadow cell, so check for 1)
new KeyValue(row, family2, qualifier1, 0, data),
- new KeyValue(row, family2, CellUtils.addShadowCellSuffix(qualifier1), 0, data)
+ new KeyValue(row, family2, CellUtils.addShadowCellSuffixPrefix(qualifier1), 0, data)
)
);
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-client/src/test/java/org/apache/omid/transaction/TestShadowCells.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/test/java/org/apache/omid/transaction/TestShadowCells.java b/hbase-client/src/test/java/org/apache/omid/transaction/TestShadowCells.java
index 6b430f7..18606a8 100644
--- a/hbase-client/src/test/java/org/apache/omid/transaction/TestShadowCells.java
+++ b/hbase-client/src/test/java/org/apache/omid/transaction/TestShadowCells.java
@@ -21,19 +21,21 @@ import com.google.common.base.Charsets;
import com.google.common.base.Optional;
import com.google.common.util.concurrent.ListenableFuture;
-import org.apache.omid.committable.CommitTable;
-import org.apache.omid.metrics.NullMetricsProvider;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.KeyValue;
+
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.omid.committable.CommitTable;
+import org.apache.omid.metrics.NullMetricsProvider;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.util.Bytes;
import org.mockito.Matchers;
import org.mockito.invocation.InvocationOnMock;
@@ -44,7 +46,6 @@ import org.testng.ITestContext;
import org.testng.annotations.Test;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
@@ -334,7 +335,11 @@ public class TestShadowCells extends OmidTestBase {
LOG.info("Waiting readAfterCommit barrier");
try {
readAfterCommit.await();
- final TTable table = spy(new TTable(hbaseConf, TEST_TABLE));
+ HTable htable = new HTable(hbaseConf, TEST_TABLE);
+ HTable healer = new HTable(hbaseConf, TEST_TABLE);
+
+ final SnapshotFilter snapshotFilter = spy(new SnapshotFilterImpl(new HTableAccessWrapper(htable, healer)));
+ final TTable table = new TTable(htable ,snapshotFilter);
doAnswer(new Answer<List<KeyValue>>() {
@SuppressWarnings("unchecked")
@Override
@@ -345,7 +350,7 @@ public class TestShadowCells extends OmidTestBase {
postCommitEnd.await();
return (List<KeyValue>) invocation.callRealMethod();
}
- }).when(table).filterCellsForSnapshot(Matchers.<List<Cell>>any(),
+ }).when(snapshotFilter).filterCellsForSnapshot(Matchers.<List<Cell>>any(),
any(HBaseTransaction.class), anyInt(), Matchers.<Map<String, Long>>any(), Matchers.<Map<String,byte[]>>any());
TransactionManager tm = newTransactionManager(context);
@@ -438,7 +443,7 @@ public class TestShadowCells extends OmidTestBase {
// delete new shadow cell
Delete del = new Delete(row2);
- del.deleteColumn(family, CellUtils.addShadowCellSuffix(qualifier));
+ del.deleteColumn(family, CellUtils.addShadowCellSuffixPrefix(qualifier));
htable.delete(del);
htable.flushCommits();
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-common/src/main/java/org/apache/omid/transaction/CellUtils.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/omid/transaction/CellUtils.java b/hbase-common/src/main/java/org/apache/omid/transaction/CellUtils.java
index c515e97..04900a6 100644
--- a/hbase-common/src/main/java/org/apache/omid/transaction/CellUtils.java
+++ b/hbase-common/src/main/java/org/apache/omid/transaction/CellUtils.java
@@ -49,9 +49,13 @@ public final class CellUtils {
private static final Logger LOG = LoggerFactory.getLogger(CellUtils.class);
static final byte[] SHADOW_CELL_SUFFIX = "\u0080".getBytes(Charsets.UTF_8); // Non printable char (128 ASCII)
- static byte[] DELETE_TOMBSTONE = Bytes.toBytes("__OMID_TOMBSTONE__");
- public static final byte[] FAMILY_DELETE_QUALIFIER = new byte[0];
+ //Prefix starts with 0 to apear before other cells in TransactionVisibilityFilter
+ static final byte[] SHADOW_CELL_PREFIX = "\u0000\u0080".getBytes(Charsets.UTF_8);
+ static byte[] DELETE_TOMBSTONE = HConstants.EMPTY_BYTE_ARRAY;
+ static byte[] LEGACY_DELETE_TOMBSTONE = Bytes.toBytes("__OMID_TOMBSTONE__");
+ public static final byte[] FAMILY_DELETE_QUALIFIER = HConstants.EMPTY_BYTE_ARRAY;
public static final String TRANSACTION_ATTRIBUTE = "__OMID_TRANSACTION__";
+ /**/
public static final String CLIENT_GET_ATTRIBUTE = "__OMID_CLIENT_GET__";
public static final String CONFLICT_FREE_MUTATION = "__OMID_CONFLICT_FREE_MUTATION__";
@@ -102,7 +106,7 @@ public final class CellUtils {
byte[] qualifier,
long version,
CellGetter cellGetter) throws IOException {
- return hasCell(row, family, addShadowCellSuffix(qualifier),
+ return hasCell(row, family, addShadowCellSuffixPrefix(qualifier),
version, cellGetter);
}
@@ -113,10 +117,12 @@ public final class CellUtils {
* @param qualLength the qualifier length
* @return the suffixed qualifier
*/
- public static byte[] addShadowCellSuffix(byte[] qualifierArray, int qualOffset, int qualLength) {
- byte[] result = new byte[qualLength + SHADOW_CELL_SUFFIX.length];
- System.arraycopy(qualifierArray, qualOffset, result, 0, qualLength);
- System.arraycopy(SHADOW_CELL_SUFFIX, 0, result, qualLength, SHADOW_CELL_SUFFIX.length);
+ public static byte[] addShadowCellSuffixPrefix(byte[] qualifierArray, int qualOffset, int qualLength) {
+ byte[] result = new byte[qualLength + SHADOW_CELL_SUFFIX.length + SHADOW_CELL_PREFIX.length];
+ System.arraycopy(SHADOW_CELL_PREFIX, 0, result,0 , SHADOW_CELL_PREFIX.length);
+ System.arraycopy(qualifierArray, qualOffset, result, SHADOW_CELL_PREFIX.length, qualLength);
+ System.arraycopy(SHADOW_CELL_SUFFIX, 0, result, qualLength + SHADOW_CELL_PREFIX.length,
+ SHADOW_CELL_SUFFIX.length);
return result;
}
@@ -127,8 +133,8 @@ public final class CellUtils {
* the qualifier to be suffixed
* @return the suffixed qualifier
*/
- public static byte[] addShadowCellSuffix(byte[] qualifier) {
- return addShadowCellSuffix(qualifier, 0, qualifier.length);
+ public static byte[] addShadowCellSuffixPrefix(byte[] qualifier) {
+ return addShadowCellSuffixPrefix(qualifier, 0, qualifier.length);
}
/**
@@ -139,12 +145,18 @@ public final class CellUtils {
* @param qualLength the qualifier length
* @return the new qualifier without the suffix
*/
- public static byte[] removeShadowCellSuffix(byte[] qualifier, int qualOffset, int qualLength) {
-
+ public static byte[] removeShadowCellSuffixPrefix(byte[] qualifier, int qualOffset, int qualLength) {
if (endsWith(qualifier, qualOffset, qualLength, SHADOW_CELL_SUFFIX)) {
- return Arrays.copyOfRange(qualifier,
- qualOffset,
- qualOffset + (qualLength - SHADOW_CELL_SUFFIX.length));
+ if (startsWith(qualifier, qualOffset,qualLength, SHADOW_CELL_PREFIX)) {
+ return Arrays.copyOfRange(qualifier,
+ qualOffset + SHADOW_CELL_PREFIX.length,
+ qualOffset + (qualLength - SHADOW_CELL_SUFFIX.length));
+ } else {
+ //support backward competatbiliy
+ return Arrays.copyOfRange(qualifier,
+ qualOffset,qualOffset + (qualLength - SHADOW_CELL_SUFFIX.length));
+ }
+
}
throw new IllegalArgumentException(
@@ -153,7 +165,7 @@ public final class CellUtils {
}
/**
- * Returns the qualifier length removing the shadow cell suffix. In case that que suffix is not found,
+ * Returns the qualifier length removing the shadow cell suffix and prefix. In case that que suffix is not found,
* just returns the length of the qualifier passed.
* @param qualifier the qualifier to remove the suffix from
* @param qualOffset the offset where the qualifier starts
@@ -163,11 +175,13 @@ public final class CellUtils {
public static int qualifierLengthFromShadowCellQualifier(byte[] qualifier, int qualOffset, int qualLength) {
if (endsWith(qualifier, qualOffset, qualLength, SHADOW_CELL_SUFFIX)) {
- return qualLength - SHADOW_CELL_SUFFIX.length;
+ if (startsWith(qualifier,qualOffset, qualLength, SHADOW_CELL_PREFIX)) {
+ return qualLength - SHADOW_CELL_SUFFIX.length - SHADOW_CELL_PREFIX.length;
+ } else {
+ return qualLength - SHADOW_CELL_SUFFIX.length;
+ }
}
-
return qualLength;
-
}
/**
@@ -203,6 +217,17 @@ public final class CellUtils {
}
/**
+ * Returns whether a cell contains a qualifier that is a delete cell
+ * column qualifier or not.
+ * @param cell the cell to check if contains the delete cell qualifier
+ * @return whether the cell passed contains a delete cell qualifier or not
+ */
+ public static boolean isFamilyDeleteCell(Cell cell) {
+ return CellUtil.matchingQualifier(cell, CellUtils.FAMILY_DELETE_QUALIFIER) &&
+ CellUtil.matchingValue(cell, HConstants.EMPTY_BYTE_ARRAY);
+ }
+
+ /**
* Returns whether a cell contains a qualifier that is a shadow cell
* column qualifier or not.
* @param cell the cell to check if contains the shadow cell qualifier
@@ -227,24 +252,26 @@ public final class CellUtils {
return result == 0;
}
+ private static boolean startsWith(byte[] value, int offset, int length, byte[] prefix) {
+ if (length <= prefix.length) {
+ return false;
+ }
+
+ int result = Bytes.compareTo(value, offset, prefix.length,
+ prefix, 0, prefix.length);
+ return result == 0;
+ }
+
/**
* Returns if a cell is marked as a tombstone.
* @param cell the cell to check
* @return whether the cell is marked as a tombstone or not
*/
public static boolean isTombstone(Cell cell) {
- return CellUtil.matchingValue(cell, DELETE_TOMBSTONE);
+ return CellUtil.matchingValue(cell, DELETE_TOMBSTONE) ||
+ CellUtil.matchingValue(cell, LEGACY_DELETE_TOMBSTONE);
}
- /**
- * Returns if a cell is a family deletion marker.
- * @param cell the cell to check
- * @return whether the cell is marked as a family deletion marker
- */
- public static boolean isFamilyDeleteCell(Cell cell) {
- return CellUtil.matchingQualifier(cell, CellUtils.FAMILY_DELETE_QUALIFIER) &&
- CellUtil.matchingValue(cell, HConstants.EMPTY_BYTE_ARRAY);
- }
/**
* Returns a new shadow cell created from a particular cell.
@@ -255,7 +282,7 @@ public final class CellUtils {
* @return the brand-new shadow cell
*/
public static Cell buildShadowCellFromCell(Cell cell, byte[] shadowCellValue) {
- byte[] shadowCellQualifier = addShadowCellSuffix(cell.getQualifierArray(),
+ byte[] shadowCellQualifier = addShadowCellSuffixPrefix(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength());
return new KeyValue(
@@ -361,8 +388,13 @@ public final class CellUtils {
int qualifierLength = qualifierLengthFromShadowCellQualifier(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength());
+ int qualifierOffset = cell.getQualifierOffset();
+ if (startsWith(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength(), SHADOW_CELL_PREFIX)) {
+ qualifierOffset = qualifierOffset + SHADOW_CELL_PREFIX.length;
+ }
if (!matchingQualifier(otherCell,
- cell.getQualifierArray(), cell.getQualifierOffset(), qualifierLength)) {
+ cell.getQualifierArray(), qualifierOffset, qualifierLength)) {
return false;
}
} else {
@@ -382,12 +414,17 @@ public final class CellUtils {
hasher.putBytes(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
hasher.putBytes(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
int qualifierLength = cell.getQualifierLength();
- if (isShadowCell()) { // Update qualifier length when qualifier is shadow cell
+ int qualifierOffset = cell.getQualifierOffset();
+ if (isShadowCell()) {
qualifierLength = qualifierLengthFromShadowCellQualifier(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength());
+ if (startsWith(cell.getQualifierArray(), cell.getQualifierOffset(),
+ cell.getQualifierLength(), SHADOW_CELL_PREFIX)) {
+ qualifierOffset = qualifierOffset + SHADOW_CELL_PREFIX.length;
+ }
}
- hasher.putBytes(cell.getQualifierArray(), cell.getQualifierOffset(), qualifierLength);
+ hasher.putBytes(cell.getQualifierArray(),qualifierOffset , qualifierLength);
hasher.putLong(cell.getTimestamp());
return hasher.hash().asInt();
}
@@ -405,7 +442,7 @@ public final class CellUtils {
cell.getQualifierOffset(),
cell.getQualifierLength());
helper.add("qualifier whithout shadow cell suffix",
- Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), qualifierLength));
+ Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset() + 1, qualifierLength));
}
helper.add("ts", cell.getTimestamp());
return helper.toString();
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/CompactorScanner.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/CompactorScanner.java b/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/CompactorScanner.java
index 12ecbb0..cf93163 100644
--- a/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/CompactorScanner.java
+++ b/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/CompactorScanner.java
@@ -235,7 +235,7 @@ public class CompactorScanner implements InternalScanner {
} else {
Get g = new Get(CellUtil.cloneRow(cell));
byte[] family = CellUtil.cloneFamily(cell);
- byte[] qualifier = CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
+ byte[] qualifier = CellUtils.addShadowCellSuffixPrefix(cell.getQualifierArray(),
cell.getQualifierOffset(),
cell.getQualifierLength());
g.addColumn(family, qualifier);
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/OmidRegionScanner.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/OmidRegionScanner.java b/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/OmidRegionScanner.java
index 7ccca00..5f7de96 100644
--- a/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/OmidRegionScanner.java
+++ b/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/OmidRegionScanner.java
@@ -18,53 +18,40 @@
package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-
+import java.util.Queue;
import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HRegionInfo;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.hadoop.hbase.regionserver.ScannerContext;
-import org.apache.omid.transaction.HBaseTransaction;
import org.apache.omid.transaction.SnapshotFilterImpl;
public class OmidRegionScanner implements RegionScanner {
private RegionScanner scanner;
private SnapshotFilterImpl snapshotFilter;
- private HBaseTransaction transaction;
- private int maxVersions;
- private Map<String, Long> familyDeletionCache;
- private Map<String,byte[]> attributeMap;
+ private final Queue<SnapshotFilterImpl> snapshotFilterQueue;
+
public OmidRegionScanner(SnapshotFilterImpl snapshotFilter,
- RegionScanner s,
- HBaseTransaction transaction,
- int maxVersions,
- Map<String,byte[]> attributeMap) {
+ RegionScanner s, Queue<SnapshotFilterImpl> snapshotFilterQueue) {
this.snapshotFilter = snapshotFilter;
this.scanner = s;
- this.transaction = transaction;
- this.maxVersions = maxVersions;
- this.familyDeletionCache = new HashMap<String, Long>();
- this.attributeMap = attributeMap;
+ this.snapshotFilterQueue = snapshotFilterQueue;
}
@Override
public boolean next(List<Cell> results) throws IOException {
- return next(results, Integer.MAX_VALUE);
+ return scanner.next(results);
}
- public boolean next(List<Cell> result, int limit) throws IOException {
- return nextRaw(result, limit);
+ @Override
+ public boolean next(List<Cell> list, ScannerContext scannerContext) throws IOException {
+ return scanner.next(list, scannerContext);
}
@Override
public void close() throws IOException {
scanner.close();
+ snapshotFilterQueue.add(snapshotFilter);
}
@Override
@@ -79,7 +66,7 @@ public class OmidRegionScanner implements RegionScanner {
@Override
public boolean reseek(byte[] row) throws IOException {
- throw new RuntimeException("Not implemented");
+ return scanner.reseek(row);
}
@Override
@@ -93,46 +80,19 @@ public class OmidRegionScanner implements RegionScanner {
}
@Override
- public boolean nextRaw(List<Cell> result) throws IOException {
- return nextRaw(result,Integer.MAX_VALUE);
- }
-
- public boolean next(List<Cell> result,
- ScannerContext scannerContext) throws IOException {
- return next(result, scannerContext.getBatchLimit());
+ public int getBatch() {
+ return scanner.getBatch();
}
- public boolean nextRaw(List<Cell> result,
- ScannerContext scannerContext) throws IOException {
- return nextRaw(result, scannerContext.getBatchLimit());
+ @Override
+ public boolean nextRaw(List<Cell> result) throws IOException {
+ return scanner.nextRaw(result);
}
- public int getBatch() {
- return Integer.MAX_VALUE;
+ @Override
+ public boolean nextRaw(List<Cell> list, ScannerContext scannerContext) throws IOException {
+ return scanner.nextRaw(list, scannerContext);
}
- public boolean nextRaw(List<Cell> result, int limit) throws IOException {
- try {
- List<Cell> filteredResult = new ArrayList<Cell>();
- while (filteredResult.isEmpty()) {
- scanner.nextRaw(filteredResult);
- if (filteredResult.isEmpty()) {
- return false;
- }
-
- filteredResult = snapshotFilter.filterCellsForSnapshot(filteredResult, transaction, maxVersions, familyDeletionCache, attributeMap);
- }
-
- for (Cell cell : filteredResult) {
- result.add(cell);
- }
-
- return true;
- } catch (IOException e) {
- throw e;
- } catch (Exception e) {
- throw new DoNotRetryIOException(e);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/RegionAccessWrapper.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/RegionAccessWrapper.java b/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/RegionAccessWrapper.java
index 28a4be3..4786eda 100644
--- a/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/RegionAccessWrapper.java
+++ b/hbase-coprocessor/src/main/java/org/apache/hadoop/hbase/regionserver/RegionAccessWrapper.java
@@ -20,10 +20,12 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.List;
+
import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
import org.apache.omid.transaction.TableAccessWrapper;
// This class wraps the Region object when doing server side filtering.
@@ -56,4 +58,9 @@ public class RegionAccessWrapper implements TableAccessWrapper {
region.put(put);
}
+ @Override
+ public ResultScanner getScanner(Scan scan) throws IOException {
+ return null;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidSnapshotFilter.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidSnapshotFilter.java b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidSnapshotFilter.java
index 2c9caff..0d13b62 100644
--- a/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidSnapshotFilter.java
+++ b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidSnapshotFilter.java
@@ -17,8 +17,13 @@
*/
package org.apache.omid.transaction;
-import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.filter.Filter;
+
+import org.apache.hadoop.hbase.regionserver.OmidRegionScanner;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.omid.committable.CommitTable;
import org.apache.omid.committable.hbase.HBaseCommitTable;
import org.apache.omid.committable.hbase.HBaseCommitTableConfig;
@@ -28,24 +33,20 @@ import org.apache.omid.HBaseShims;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
-import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.hadoop.hbase.regionserver.OmidRegionScanner;
import org.apache.hadoop.hbase.regionserver.RegionAccessWrapper;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
+
import java.util.HashSet;
import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import static org.apache.omid.committable.hbase.HBaseCommitTableConfig.COMMIT_TABLE_NAME_KEY;
@@ -58,22 +59,19 @@ public class OmidSnapshotFilter extends BaseRegionObserver {
private HBaseCommitTableConfig commitTableConf = null;
private Configuration conf = null;
- @VisibleForTesting
- private CommitTable.Client commitTableClient;
+ Queue<SnapshotFilterImpl> snapshotFilterQueue = new ConcurrentLinkedQueue<>();
- private SnapshotFilterImpl snapshotFilter;
+ private CommitTable.Client inMemoryCommitTable = null;
- final static String OMID_SNAPSHOT_FILTER_CF_FLAG = "OMID_SNAPSHOT_FILTER_ENABLED";
+ public OmidSnapshotFilter(CommitTable.Client commitTableClient) {
+ LOG.info("Compactor coprocessor initialized with constructor for testing");
+ this.inMemoryCommitTable = commitTableClient;
+ }
public OmidSnapshotFilter() {
LOG.info("Compactor coprocessor initialized via empty constructor");
}
- public OmidSnapshotFilter(CommitTable.Client commitTableClient) {
- LOG.info("Compactor coprocessor initialized with constructor for testing");
- this.commitTableClient = commitTableClient;
- }
-
@Override
public void start(CoprocessorEnvironment env) throws IOException {
LOG.info("Starting snapshot filter coprocessor");
@@ -83,92 +81,99 @@ public class OmidSnapshotFilter extends BaseRegionObserver {
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
+ 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());
+ }
+ }
- snapshotFilter.setTableAccessWrapper(regionAccessWrapper);
- List<Cell> filteredKeyValues = Collections.emptyList();
- if (!res.isEmpty()) {
- TSOProto.Transaction transaction = TSOProto.Transaction.parseFrom(get.getAttribute(CellUtils.TRANSACTION_ATTRIBUTE));
+ @Override
+ public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get, List<Cell> results)
+ throws IOException {
- long id = transaction.getTimestamp();
- long readTs = transaction.getReadTimestamp();
- long epoch = transaction.getEpoch();
- VisibilityLevel visibilityLevel = VisibilityLevel.fromInteger(transaction.getVisibilityLevel());
+ if (get.getAttribute(CellUtils.CLIENT_GET_ATTRIBUTE) == null) return;
+ HBaseTransaction hbaseTransaction = getHBaseTransaction(get.getAttribute(CellUtils.TRANSACTION_ATTRIBUTE));
+ SnapshotFilterImpl snapshotFilter = getSnapshotFilter(e);
- 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());
- }
+ // In order to get hbase FilterBase framework to keep getting more versions
+ get.setMaxVersions();
+ Filter newFilter = TransactionFilters.getVisibilityFilter(get.getFilter(),
+ snapshotFilter, hbaseTransaction);
+ get.setFilter(newFilter);
+ }
- for (Cell cell : filteredKeyValues) {
- result.add(cell);
- }
- c.bypass();
- } catch (IOException e) {
- throw e;
- } catch (Exception e) {
- throw new DoNotRetryIOException(e);
+ private SnapshotFilterImpl getSnapshotFilter(ObserverContext<RegionCoprocessorEnvironment> e)
+ throws IOException {
+ SnapshotFilterImpl snapshotFilter= snapshotFilterQueue.poll();
+ if (snapshotFilter == null) {
+ RegionAccessWrapper regionAccessWrapper =
+ new RegionAccessWrapper(HBaseShims.getRegionCoprocessorRegion(e.getEnvironment()));
+ snapshotFilter = new SnapshotFilterImpl(regionAccessWrapper, initAndGetCommitTableClient());
}
+ return snapshotFilter;
}
+
@Override
- public RegionScanner postScannerOpen(ObserverContext<RegionCoprocessorEnvironment> e,
- Scan scan,
- RegionScanner s) throws IOException {
+ public RegionScanner preScannerOpen(ObserverContext<RegionCoprocessorEnvironment> e,
+ Scan scan,
+ RegionScanner s) throws IOException {
byte[] byteTransaction = scan.getAttribute(CellUtils.TRANSACTION_ATTRIBUTE);
if (byteTransaction == null) {
return s;
}
- TSOProto.Transaction transaction = TSOProto.Transaction.parseFrom(byteTransaction);
+ HBaseTransaction hbaseTransaction = getHBaseTransaction(byteTransaction);
+ SnapshotFilterImpl snapshotFilter = getSnapshotFilter(e);
+
+ scan.setMaxVersions();
+ Filter newFilter = TransactionFilters.getVisibilityFilter(scan.getFilter(),
+ snapshotFilter, hbaseTransaction);
+ scan.setFilter(newFilter);
+ return new OmidRegionScanner(snapshotFilter, s, snapshotFilterQueue);
+ }
+
+ private HBaseTransaction getHBaseTransaction(byte[] byteTransaction)
+ throws InvalidProtocolBufferException {
+ TSOProto.Transaction transaction = TSOProto.Transaction.parseFrom(byteTransaction);
long id = transaction.getTimestamp();
long readTs = transaction.getReadTimestamp();
long epoch = transaction.getEpoch();
VisibilityLevel visibilityLevel = VisibilityLevel.fromInteger(transaction.getVisibilityLevel());
- HBaseTransaction hbaseTransaction = new HBaseTransaction(id, readTs, visibilityLevel, epoch, new HashSet<HBaseCellId>(), new HashSet<HBaseCellId>(), null);
-
- RegionAccessWrapper regionAccessWrapper = new RegionAccessWrapper(HBaseShims.getRegionCoprocessorRegion(e.getEnvironment()));
-
- snapshotFilter.setTableAccessWrapper(regionAccessWrapper);
-
- return new OmidRegionScanner(snapshotFilter, s, hbaseTransaction, 1, scan.getAttributesMap());
+ return new HBaseTransaction(id, readTs, visibilityLevel, epoch, new HashSet<HBaseCellId>(), new HashSet<HBaseCellId>(), null);
}
private CommitTable.Client initAndGetCommitTableClient() throws IOException {
LOG.info("Trying to get the commit table client");
+ if (inMemoryCommitTable != null) {
+ return inMemoryCommitTable;
+ }
CommitTable commitTable = new HBaseCommitTable(conf, commitTableConf);
CommitTable.Client commitTableClient = commitTable.getClient();
LOG.info("Commit table client obtained {}", commitTableClient.getClass().getCanonicalName());
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/395fb46f/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionFilters.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionFilters.java b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionFilters.java
new file mode 100644
index 0000000..ae7516c
--- /dev/null
+++ b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/TransactionFilters.java
@@ -0,0 +1,30 @@
+/*
+ * 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 com.beust.jcommander.internal.Nullable;
+import org.apache.hadoop.hbase.filter.Filter;
+
+public class TransactionFilters {
+
+ public static Filter getVisibilityFilter(@Nullable Filter cellFilter,
+ SnapshotFilterImpl regionAccessWrapper,
+ HBaseTransaction hbaseTransaction) {
+ return new TransactionVisibilityFilter(cellFilter, regionAccessWrapper, hbaseTransaction);
+ }
+}