You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@omid.apache.org by fp...@apache.org on 2018/01/31 00:00:48 UTC
[4/4] incubator-omid git commit: [OMID-79] Revert to the state of the
project at 0c371361781957c96b20295290a167f7be3b33e2
[OMID-79] Revert to the state of the project at 0c371361781957c96b20295290a167f7be3b33e2
This reverts the state of the master branch to the state it was before Ohad's commits related to
Phoenix integration:
https://github.com/apache/incubator-omid/commit/0005b3603571b28e01e5fbd8943e23efc5ead781
https://github.com/apache/incubator-omid/commit/60c9e714fd862f17407202565ac1eb7011910e1e
https://github.com/apache/incubator-omid/commit/0cae2ff18f507f640edbe484e24000630856d3f0
https://github.com/apache/incubator-omid/commit/1f83aeda963472c861b4b622c08b3c8c725fbd16
https://github.com/apache/incubator-omid/commit/4c8d7821fbbad943f46f3dfc0ac51b0ea908f90f
https://github.com/apache/incubator-omid/commit/fab2cfebbd9f4c3d95a1d02a428c90c1dc649db5
Change-Id: Id94c40909e1522dbe01861eb1ccfef47fb4c0668
Project: http://git-wip-us.apache.org/repos/asf/incubator-omid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-omid/commit/31dd269d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-omid/tree/31dd269d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-omid/diff/31dd269d
Branch: refs/heads/master
Commit: 31dd269dc88f9617ac37e8dfb5fd867db389ba7c
Parents: fab2cfe
Author: Francisco Perez-Sorrosal <fp...@apache.org>
Authored: Tue Jan 30 15:36:14 2018 -0800
Committer: Francisco Perez-Sorrosal <fp...@apache.org>
Committed: Tue Jan 30 16:00:07 2018 -0800
----------------------------------------------------------------------
common/src/main/proto/TSOProto.proto | 19 -
.../transaction/AttributeSetSnapshotFilter.java | 85 ---
.../apache/omid/transaction/HBaseCellId.java | 22 +-
.../HBaseOmidClientConfiguration.java | 10 -
.../transaction/HBaseSyncPostCommitter.java | 2 +-
.../omid/transaction/HBaseTransaction.java | 10 +-
.../transaction/HBaseTransactionClient.java | 2 +
.../transaction/HBaseTransactionManager.java | 47 +-
.../omid/transaction/HTableAccessWrapper.java | 54 --
.../apache/omid/transaction/SnapshotFilter.java | 50 --
.../omid/transaction/SnapshotFilterImpl.java | 508 ------------------
.../org/apache/omid/transaction/TTable.java | 345 ++++++++-----
.../omid/transaction/TableAccessWrapper.java | 34 --
.../apache/omid/transaction/TestCheckpoint.java | 320 ------------
.../omid/transaction/TestColumnIterator.java | 2 +-
.../transaction/TestHBaseTransactionClient.java | 60 ++-
.../TestHBaseTransactionManager.java | 2 +-
.../omid/transaction/TestShadowCells.java | 7 +-
.../org/apache/omid/transaction/CellUtils.java | 3 -
hbase-coprocessor/pom.xml | 1 +
.../hbase/regionserver/OmidRegionScanner.java | 128 -----
.../hbase/regionserver/RegionAccessWrapper.java | 59 ---
.../omid/transaction/OmidSnapshotFilter.java | 160 ------
.../TSOForSnapshotFilterTestModule.java | 134 -----
.../omid/transaction/TestSnapshotFilter.java | 514 -------------------
.../hadoop/hbase/regionserver/Region.java | 7 -
.../omid/transaction/AbstractTransaction.java | 108 +---
.../transaction/AbstractTransactionManager.java | 140 +++--
.../omid/transaction/TransactionManager.java | 11 -
.../java/org/apache/omid/tso/client/CellId.java | 3 -
.../apache/omid/tso/client/MockTSOClient.java | 68 +--
.../tso/client/OmidClientConfiguration.java | 12 -
.../org/apache/omid/tso/client/TSOClient.java | 129 +----
.../org/apache/omid/tso/client/TSOProtocol.java | 14 -
.../apache/omid/tso/util/DummyCellIdImpl.java | 15 -
.../src/main/resources/omid-client-config.yml | 6 +-
tso-server/pom.xml | 1 +
.../main/java/org/apache/omid/tso/Batch.java | 10 -
.../java/org/apache/omid/tso/PersistEvent.java | 12 +-
.../apache/omid/tso/PersistenceProcessor.java | 2 -
.../omid/tso/PersistenceProcessorHandler.java | 9 -
.../omid/tso/PersistenceProcessorImpl.java | 18 +-
.../org/apache/omid/tso/ReplyProcessor.java | 13 -
.../org/apache/omid/tso/ReplyProcessorImpl.java | 19 -
.../org/apache/omid/tso/RequestProcessor.java | 3 +-
.../apache/omid/tso/RequestProcessorImpl.java | 121 +----
.../org/apache/omid/tso/TSOChannelHandler.java | 4 -
.../java/org/apache/omid/tso/TSOModule.java | 11 +-
.../org/apache/omid/tso/TSOServerConfig.java | 19 -
.../apache/omid/tso/TimestampOracleImpl.java | 10 +-
.../apache/omid/tso/WorldClockOracleImpl.java | 185 -------
.../default-omid-server-configuration.yml | 5 +-
.../apache/omid/tso/TestRequestProcessor.java | 42 +-
.../omid/tso/TestTSOChannelHandlerNetty.java | 22 +-
.../apache/omid/tso/TestTimestampOracle.java | 5 +-
.../apache/omid/tso/TestWorldTimeOracle.java | 119 -----
...tionOfTSOClientServerBasicFunctionality.java | 58 +--
.../client/TestTSOClientConnectionToTSO.java | 8 +-
...stTSOClientRequestAndResponseBehaviours.java | 4 +-
.../TestTSOClientRowAndCellLevelConflict.java | 203 --------
60 files changed, 461 insertions(+), 3533 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/common/src/main/proto/TSOProto.proto
----------------------------------------------------------------------
diff --git a/common/src/main/proto/TSOProto.proto b/common/src/main/proto/TSOProto.proto
index b434421..749beaa 100644
--- a/common/src/main/proto/TSOProto.proto
+++ b/common/src/main/proto/TSOProto.proto
@@ -24,7 +24,6 @@ message Request {
optional TimestampRequest timestampRequest = 1;
optional CommitRequest commitRequest = 2;
optional HandshakeRequest handshakeRequest = 3;
- optional FenceRequest fenceRequest = 4;
}
message TimestampRequest {
@@ -34,29 +33,18 @@ message CommitRequest {
optional int64 startTimestamp = 1;
optional bool isRetry = 2 [default = false];
repeated int64 cellId = 3;
- repeated int64 TableId = 4;
-}
-
-message FenceRequest {
- optional int64 TableId = 1;
}
message Response {
optional TimestampResponse timestampResponse = 1;
optional CommitResponse commitResponse = 2;
optional HandshakeResponse handshakeResponse = 3;
- optional FenceResponse fenceResponse = 4;
}
message TimestampResponse {
optional int64 startTimestamp = 1;
}
-message FenceResponse {
- optional int64 TableId = 1;
- optional int64 FenceId = 2;
-}
-
message CommitResponse {
optional bool aborted = 1;
optional int64 startTimestamp = 2;
@@ -76,10 +64,3 @@ message HandshakeResponse {
optional bool clientCompatible = 1;
optional Capabilities serverCapabilities = 2;
}
-
-message Transaction {
- optional int64 timestamp = 1;
- optional int64 readTimestamp = 2;
- optional int32 visibilityLevel = 3;
- optional int64 epoch = 4;
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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
deleted file mode 100644
index 6906939..0000000
--- a/hbase-client/src/main/java/org/apache/omid/transaction/AttributeSetSnapshotFilter.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * 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 java.io.IOException;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.client.Get;
-import org.apache.hadoop.hbase.client.HTableInterface;
-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.util.Bytes;
-import org.apache.omid.committable.CommitTable.CommitTimestamp;
-import org.apache.omid.proto.TSOProto;
-
-import com.google.common.base.Optional;
-
-public class AttributeSetSnapshotFilter implements SnapshotFilter {
-
- private HTableInterface table;
-
- public AttributeSetSnapshotFilter(HTableInterface table) {
- this.table = table;
- }
-
- private TSOProto.Transaction.Builder getBuilder(HBaseTransaction transaction) {
- return TSOProto.Transaction.newBuilder().setTimestamp(transaction.getTransactionId())
- .setReadTimestamp(transaction.getReadTimestamp())
- .setVisibilityLevel(transaction.getVisibilityLevel().ordinal())
- .setEpoch(transaction.getEpoch());
- }
-
- @Override
- public Result get(TTable ttable, Get get, HBaseTransaction transaction) throws IOException {
- get.setAttribute(CellUtils.TRANSACTION_ATTRIBUTE, getBuilder(transaction).build().toByteArray());
- get.setAttribute(CellUtils.CLIENT_GET_ATTRIBUTE, Bytes.toBytes(true));
-
- return table.get(get);
- }
-
- @Override
- public ResultScanner getScanner(TTable ttable, Scan scan, HBaseTransaction transaction) throws IOException {
- scan.setAttribute(CellUtils.TRANSACTION_ATTRIBUTE, getBuilder(transaction).build().toByteArray());
-
- return table.getScanner(scan);
- }
-
- @Override
- public List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
- int versionsToRequest, Map<String, List<Cell>> familyDeletionCache) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- public boolean isCommitted(HBaseCellId hBaseCellId, long epoch) throws TransactionException {
- throw new UnsupportedOperationException();
- }
-
- public CommitTimestamp locateCellCommitTimestamp(long cellStartTimestamp, long epoch,
- CommitTimestampLocator locator) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- public Optional<CommitTimestamp> readCommitTimestampFromShadowCell(long cellStartTimestamp, CommitTimestampLocator locator)
- throws IOException {
- throw new UnsupportedOperationException();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java
index 63e6376..8d0641b 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseCellId.java
@@ -17,9 +17,7 @@
*/
package org.apache.omid.transaction;
-import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
-
import org.apache.omid.tso.client.CellId;
import org.apache.hadoop.hbase.client.HTableInterface;
@@ -71,7 +69,7 @@ public class HBaseCellId implements CellId {
@Override
public long getCellId() {
- return getHasher()
+ return Hashing.murmur3_128().newHasher()
.putBytes(table.getTableName())
.putBytes(row)
.putBytes(family)
@@ -79,22 +77,4 @@ public class HBaseCellId implements CellId {
.hash().asLong();
}
- @Override
- public long getTableId() {
- return getHasher()
- .putBytes(table.getTableName())
- .hash().asLong();
- }
-
- @Override
- public long getRowId() {
- return getHasher()
- .putBytes(table.getTableName())
- .putBytes(row)
- .hash().asLong();
- }
-
- public static Hasher getHasher() {
- return Hashing.murmur3_128().newHasher();
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java
index d945688..f5a1823 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseOmidClientConfiguration.java
@@ -20,11 +20,9 @@ package org.apache.omid.transaction;
import com.google.common.annotations.VisibleForTesting;
import com.google.inject.Inject;
import com.google.inject.name.Named;
-
import org.apache.omid.YAMLUtils;
import org.apache.omid.metrics.MetricsRegistry;
import org.apache.omid.tools.hbase.SecureHBaseConfig;
-import org.apache.omid.tso.client.OmidClientConfiguration.ConflictDetectionLevel;
import org.apache.omid.tso.client.OmidClientConfiguration.PostCommitMode;
import org.apache.omid.tso.client.OmidClientConfiguration;
import org.apache.hadoop.conf.Configuration;
@@ -75,14 +73,6 @@ public class HBaseOmidClientConfiguration extends SecureHBaseConfig {
omidClientConfiguration.setPostCommitMode(postCommitMode);
}
- public ConflictDetectionLevel getConflictAnalysisLevel() {
- return omidClientConfiguration.getConflictAnalysisLevel();
- }
-
- public void setConflictAnalysisLevel(ConflictDetectionLevel conflictAnalysisLevel) {
- omidClientConfiguration.setConflictAnalysisLevel(conflictAnalysisLevel);
- }
-
public String getCommitTableName() {
return commitTableName;
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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 06e5c89..952d067 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
@@ -66,7 +66,7 @@ public class HBaseSyncPostCommitter implements PostCommitActions {
Put put = new Put(cell.getRow());
put.add(cell.getFamily(),
CellUtils.addShadowCellSuffix(cell.getQualifier(), 0, cell.getQualifier().length),
- cell.getTimestamp(),
+ tx.getStartTimestamp(),
Bytes.toBytes(tx.getCommitTimestamp()));
try {
cell.getTable().put(put);
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java
index c32aff7..2bae5f5 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransaction.java
@@ -19,32 +19,26 @@ package org.apache.omid.transaction;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.omid.transaction.AbstractTransaction.VisibilityLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.HashSet;
-import java.util.List;
import java.util.Set;
public class HBaseTransaction extends AbstractTransaction<HBaseCellId> {
private static final Logger LOG = LoggerFactory.getLogger(HBaseTransaction.class);
- public HBaseTransaction(long transactionId, long epoch, Set<HBaseCellId> writeSet, AbstractTransactionManager tm) {
+ HBaseTransaction(long transactionId, long epoch, Set<HBaseCellId> writeSet, AbstractTransactionManager tm) {
super(transactionId, epoch, writeSet, tm);
}
- public HBaseTransaction(long transactionId, long readTimestamp, VisibilityLevel visibilityLevel, long epoch, Set<HBaseCellId> writeSet, AbstractTransactionManager tm) {
- super(transactionId, readTimestamp, visibilityLevel, epoch, writeSet, tm);
- }
-
@Override
public void cleanup() {
Set<HBaseCellId> writeSet = getWriteSet();
for (final HBaseCellId cell : writeSet) {
Delete delete = new Delete(cell.getRow());
- delete.deleteColumn(cell.getFamily(), cell.getQualifier(), cell.getTimestamp());
+ delete.deleteColumn(cell.getFamily(), cell.getQualifier(), getStartTimestamp());
try {
cell.getTable().delete(delete);
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java
index f5311fb..c5d4d59 100644
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java
+++ b/hbase-client/src/main/java/org/apache/omid/transaction/HBaseTransactionClient.java
@@ -18,5 +18,7 @@
package org.apache.omid.transaction;
public interface HBaseTransactionClient {
+ boolean isCommitted(HBaseCellId hBaseCellId) throws TransactionException;
+
long getLowWatermark() throws TransactionException;
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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 71d1f98..990758c 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
@@ -23,14 +23,12 @@ 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;
import org.apache.omid.tso.client.CellId;
-import org.apache.omid.tso.client.OmidClientConfiguration.ConflictDetectionLevel;
import org.apache.omid.tso.client.TSOClient;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
@@ -197,9 +195,35 @@ public class HBaseTransactionManager extends AbstractTransactionManager implemen
}
}
+ // ----------------------------------------------------------------------------------------------------------------
+ // HBaseTransactionClient method implementations
+ // ----------------------------------------------------------------------------------------------------------------
+
@Override
- public long getHashForTable(byte[] tableName) {
- return HBaseCellId.getHasher().putBytes(tableName).hash().asLong();
+ public boolean isCommitted(HBaseCellId hBaseCellId) throws TransactionException {
+ try {
+ CommitTimestamp tentativeCommitTimestamp =
+ locateCellCommitTimestamp(hBaseCellId.getTimestamp(), tsoClient.getEpoch(),
+ new CommitTimestampLocatorImpl(hBaseCellId, Maps.<Long, Long>newHashMap()));
+
+ // If transaction that added the cell was invalidated
+ if (!tentativeCommitTimestamp.isValid()) {
+ return false;
+ }
+
+ switch (tentativeCommitTimestamp.getLocation()) {
+ case COMMIT_TABLE:
+ case SHADOW_CELL:
+ return true;
+ case NOT_PRESENT:
+ return false;
+ case CACHE: // cache was empty
+ default:
+ return false;
+ }
+ } catch (IOException e) {
+ throw new TransactionException("Failure while checking if a transaction was committed", e);
+ }
}
@Override
@@ -229,27 +253,14 @@ public class HBaseTransactionManager extends AbstractTransactionManager implemen
}
- public ConflictDetectionLevel getConflictDetectionLevel() {
- return tsoClient.getConflictDetectionLevel();
- }
-
static class CommitTimestampLocatorImpl implements CommitTimestampLocator {
private HBaseCellId hBaseCellId;
private final Map<Long, Long> commitCache;
- private TableAccessWrapper tableAccessWrapper;
-
- CommitTimestampLocatorImpl(HBaseCellId hBaseCellId, Map<Long, Long> commitCache, TableAccessWrapper tableAccessWrapper) {
- this.hBaseCellId = hBaseCellId;
- this.commitCache = commitCache;
- this.tableAccessWrapper = tableAccessWrapper;
- }
CommitTimestampLocatorImpl(HBaseCellId hBaseCellId, Map<Long, Long> commitCache) {
this.hBaseCellId = hBaseCellId;
this.commitCache = commitCache;
- this.tableAccessWrapper = null;
- this.tableAccessWrapper = new HTableAccessWrapper(hBaseCellId.getTable(), hBaseCellId.getTable());
}
@Override
@@ -269,7 +280,7 @@ public class HBaseTransactionManager extends AbstractTransactionManager implemen
get.addColumn(family, shadowCellQualifier);
get.setMaxVersions(1);
get.setTimeStamp(startTimestamp);
- Result result = tableAccessWrapper.get(get);
+ Result result = hBaseCellId.getTable().get(get);
if (result.containsColumn(family, shadowCellQualifier)) {
return Optional.of(Bytes.toLong(result.getValue(family, shadowCellQualifier)));
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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
deleted file mode 100644
index 1994cad..0000000
--- a/hbase-client/src/main/java/org/apache/omid/transaction/HTableAccessWrapper.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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 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;
-
-// This class wraps the HTableInterface object when doing client side filtering.
-public class HTableAccessWrapper implements TableAccessWrapper {
-
- private final HTableInterface writeTable;
- private final HTableInterface readTable;
-
- public HTableAccessWrapper(HTableInterface table, HTableInterface healerTable) {
- this.readTable = table;
- this.writeTable = healerTable;
- }
-
- @Override
- public Result[] get(List<Get> get) throws IOException {
- return readTable.get(get);
- }
-
- @Override
- public Result get(Get get) throws IOException {
- return readTable.get(get);
- }
-
- @Override
- public void put(Put put) throws IOException {
- writeTable.put(put);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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
deleted file mode 100644
index 112544f..0000000
--- a/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilter.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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 java.io.IOException;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.client.Get;
-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.CommitTimestamp;
-
-import com.google.common.base.Optional;
-
-public interface SnapshotFilter {
-
- public Result get(TTable ttable, Get get, HBaseTransaction transaction) throws IOException;
-
- public ResultScanner getScanner(TTable ttable, Scan scan, HBaseTransaction transaction) throws IOException;
-
- public List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
- int versionsToRequest, Map<String, List<Cell>> familyDeletionCache) throws IOException;
-
- public boolean isCommitted(HBaseCellId hBaseCellId, long epoch) throws TransactionException;
-
- public CommitTimestamp locateCellCommitTimestamp(long cellStartTimestamp, long epoch,
- CommitTimestampLocator locator) throws IOException;
-
- public Optional<CommitTimestamp> readCommitTimestampFromShadowCell(long cellStartTimestamp, CommitTimestampLocator locator)
- throws IOException;
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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
deleted file mode 100644
index a656aec..0000000
--- a/hbase-client/src/main/java/org/apache/omid/transaction/SnapshotFilterImpl.java
+++ /dev/null
@@ -1,508 +0,0 @@
-/*
- * 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 static org.apache.omid.committable.CommitTable.CommitTimestamp.Location.CACHE;
-import static org.apache.omid.committable.CommitTable.CommitTimestamp.Location.COMMIT_TABLE;
-import static org.apache.omid.committable.CommitTable.CommitTimestamp.Location.NOT_PRESENT;
-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.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;
-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;
-
-import com.google.common.base.Function;
-import com.google.common.base.Optional;
-import com.google.common.base.Predicate;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Multimaps;
-
-public class SnapshotFilterImpl implements SnapshotFilter {
-
- private static Logger LOG = LoggerFactory.getLogger(SnapshotFilterImpl.class);
-
- private TableAccessWrapper tableAccessWrapper;
-
- private CommitTable.Client commitTableClient;
-
- public SnapshotFilterImpl(TableAccessWrapper tableAccessWrapper, CommitTable.Client commitTableClient) throws IOException {
- this.tableAccessWrapper = tableAccessWrapper;
- this.commitTableClient = commitTableClient;
- }
-
- public SnapshotFilterImpl(TableAccessWrapper tableAccessWrapper) throws IOException {
- this(tableAccessWrapper, null);
- }
-
- public SnapshotFilterImpl(CommitTable.Client commitTableClient) throws IOException {
- this(null, commitTableClient);
- }
-
- void setTableAccessWrapper(TableAccessWrapper tableAccessWrapper) {
- this.tableAccessWrapper = tableAccessWrapper;
- }
-
- /**
- * Check whether a cell was deleted using family deletion marker
- *
- * @param cell The cell to check
- * @param transaction Defines the current snapshot
- * @param familyDeletionCache Accumulates the family deletion markers to identify cells that deleted with a higher version
- * @param commitCache Holds shadow cells information
- * @return Whether the cell was deleted
- */
- private boolean checkFamilyDeletionCache(Cell cell, HBaseTransaction transaction, Map<String, List<Cell>> familyDeletionCache, Map<Long, Long> commitCache) throws IOException {
- List<Cell> familyDeletionCells = familyDeletionCache.get(Bytes.toString((cell.getRow())));
- if (familyDeletionCells != null) {
- for(Cell familyDeletionCell : familyDeletionCells) {
- String family = Bytes.toString(cell.getFamily());
- String familyDeletion = Bytes.toString(familyDeletionCell.getFamily());
- if (family.equals(familyDeletion)) {
- Optional<Long> familyDeletionCommitTimestamp = getCommitTimestamp(familyDeletionCell, transaction, commitCache);
- if (familyDeletionCommitTimestamp.isPresent() && familyDeletionCommitTimestamp.get() >= 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(),
- cell.getQualifierOffset(),
- cell.getQualifierLength());
- put.add(family, shadowCellQualifier, cell.getTimestamp(), Bytes.toBytes(commitTimestamp));
- try {
- tableAccessWrapper.put(put);
- } catch (IOException e) {
- LOG.warn("Failed healing shadow cell for kv {}", cell, e);
- }
- }
-
- /**
- * Check if the transaction commit data is in the shadow cell
- * @param cellStartTimestamp
- * the transaction start timestamp
- * locator
- * the timestamp locator
- * @throws IOException
- */
- @Override
- public Optional<CommitTimestamp> readCommitTimestampFromShadowCell(long cellStartTimestamp, CommitTimestampLocator locator)
- throws IOException
- {
-
- Optional<CommitTimestamp> commitTS = Optional.absent();
-
- Optional<Long> commitTimestamp = locator.readCommitTimestampFromShadowCell(cellStartTimestamp);
- if (commitTimestamp.isPresent()) {
- commitTS = Optional.of(new CommitTimestamp(SHADOW_CELL, commitTimestamp.get(), true)); // Valid commit TS
- }
-
- return commitTS;
- }
-
- /**
- * This function returns the commit timestamp for a particular cell if the transaction was already committed in
- * the system. In case the transaction was not committed and the cell was written by transaction initialized by a
- * previous TSO server, an invalidation try occurs.
- * Otherwise the function returns a value that indicates that the commit timestamp was not found.
- * @param cellStartTimestamp
- * start timestamp of the cell to locate the commit timestamp for.
- * @param epoch
- * the epoch of the TSO server the current tso client is working with.
- * @param locator
- * a locator to find the commit timestamp in the system.
- * @return the commit timestamp joint with the location where it was found
- * or an object indicating that it was not found in the system
- * @throws IOException in case of any I/O issues
- */
- @Override
- public CommitTimestamp locateCellCommitTimestamp(long cellStartTimestamp, long epoch,
- CommitTimestampLocator locator) throws IOException {
-
- try {
- // 1) First check the cache
- Optional<Long> commitTimestamp = locator.readCommitTimestampFromCache(cellStartTimestamp);
- if (commitTimestamp.isPresent()) { // Valid commit timestamp
- return new CommitTimestamp(CACHE, commitTimestamp.get(), true);
- }
-
- // 2) Then check the commit table
- // If the data was written at a previous epoch, check whether the transaction was invalidated
- Optional<CommitTimestamp> commitTimeStamp = commitTableClient.getCommitTimestamp(cellStartTimestamp).get();
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // 3) Read from shadow cell
- commitTimeStamp = readCommitTimestampFromShadowCell(cellStartTimestamp, locator);
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // 4) Check the epoch and invalidate the entry
- // if the data was written by a transaction from a previous epoch (previous TSO)
- if (cellStartTimestamp < epoch) {
- boolean invalidated = commitTableClient.tryInvalidateTransaction(cellStartTimestamp).get();
- if (invalidated) { // Invalid commit timestamp
- return new CommitTimestamp(COMMIT_TABLE, CommitTable.INVALID_TRANSACTION_MARKER, false);
- }
- }
-
- // 5) We did not manage to invalidate the transactions then check the commit table
- commitTimeStamp = commitTableClient.getCommitTimestamp(cellStartTimestamp).get();
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // 6) Read from shadow cell
- commitTimeStamp = readCommitTimestampFromShadowCell(cellStartTimestamp, locator);
- if (commitTimeStamp.isPresent()) {
- return commitTimeStamp.get();
- }
-
- // *) Otherwise return not found
- return new CommitTimestamp(NOT_PRESENT, -1L /** TODO Check if we should return this */, true);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IOException("Interrupted while finding commit timestamp", e);
- } catch (ExecutionException e) {
- throw new IOException("Problem finding commit timestamp", e);
- }
-
- }
-
- private Optional<Long> tryToLocateCellCommitTimestamp(long epoch,
- Cell cell,
- Map<Long, Long> commitCache)
- throws IOException {
-
- CommitTimestamp tentativeCommitTimestamp =
- locateCellCommitTimestamp(
- cell.getTimestamp(),
- epoch,
- new CommitTimestampLocatorImpl(
- new HBaseCellId(null,
- CellUtil.cloneRow(cell),
- CellUtil.cloneFamily(cell),
- CellUtil.cloneQualifier(cell),
- cell.getTimestamp()),
- commitCache,
- tableAccessWrapper));
-
- // If transaction that added the cell was invalidated
- if (!tentativeCommitTimestamp.isValid()) {
- return Optional.absent();
- }
-
- switch (tentativeCommitTimestamp.getLocation()) {
- case COMMIT_TABLE:
- // If the commit timestamp is found in the persisted commit table,
- // that means the writing process of the shadow cell in the post
- // commit phase of the client probably failed, so we heal the shadow
- // cell with the right commit timestamp for avoiding further reads to
- // hit the storage
- healShadowCell(cell, tentativeCommitTimestamp.getValue());
- return Optional.of(tentativeCommitTimestamp.getValue());
- case CACHE:
- case SHADOW_CELL:
- return Optional.of(tentativeCommitTimestamp.getValue());
- case NOT_PRESENT:
- return Optional.absent();
- default:
- assert (false);
- return Optional.absent();
- }
- }
-
-
- private Optional<Long> getCommitTimestamp(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache)
- throws IOException {
-
- long startTimestamp = transaction.getStartTimestamp();
-
- if (kv.getTimestamp() == startTimestamp) {
- return Optional.of(startTimestamp);
- }
-
- if (commitTableClient == null) {
- assert (transaction.getTransactionManager() != null);
- commitTableClient = transaction.getTransactionManager().getCommitTableClient();
- }
-
- return tryToLocateCellCommitTimestamp(transaction.getEpoch(), kv,
- commitCache);
- }
-
- private Map<Long, Long> buildCommitCache(List<Cell> rawCells) {
-
- Map<Long, Long> commitCache = new HashMap<>();
-
- for (Cell cell : rawCells) {
- if (CellUtils.isShadowCell(cell)) {
- commitCache.put(cell.getTimestamp(), Bytes.toLong(CellUtil.cloneValue(cell)));
- }
- }
-
- return commitCache;
- }
-
- private void buildFamilyDeletionCache(List<Cell> rawCells, Map<String, List<Cell>> familyDeletionCache) {
-
- for (Cell cell : rawCells) {
- if (CellUtil.matchingQualifier(cell, CellUtils.FAMILY_DELETE_QUALIFIER) &&
- CellUtil.matchingValue(cell, HConstants.EMPTY_BYTE_ARRAY)) {
-
- String row = Bytes.toString(cell.getRow());
- List<Cell> cells = familyDeletionCache.get(row);
- if (cells == null) {
- cells = new ArrayList<>();
- familyDeletionCache.put(row, cells);
- }
-
- cells.add(cell);
- }
- }
-
- }
-
- private boolean isCellInTransaction(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache) {
-
- long startTimestamp = transaction.getStartTimestamp();
- long readTimestamp = transaction.getReadTimestamp();
-
- // A cell was written by a transaction if its timestamp is larger than its startTimestamp and smaller or equal to its readTimestamp.
- // There also might be a case where the cell was written by the transaction and its timestamp equals to its writeTimestamp, however,
- // this case occurs after checkpoint and in this case we do not want to read this data.
- if (kv.getTimestamp() >= startTimestamp && kv.getTimestamp() <= readTimestamp) {
- return true;
- }
-
- return false;
- }
-
- private boolean isCellInSnapshot(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache)
- throws IOException {
-
- Optional<Long> commitTimestamp = getCommitTimestamp(kv, transaction, commitCache);
-
- return commitTimestamp.isPresent() && commitTimestamp.get() < transaction.getStartTimestamp();
- }
-
- private Get createPendingGet(Cell cell, int versionCount) throws IOException {
-
- Get pendingGet = new Get(CellUtil.cloneRow(cell));
- pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell));
- pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
- cell.getQualifierOffset(),
- cell.getQualifierLength()));
- pendingGet.setMaxVersions(versionCount);
- pendingGet.setTimeRange(0, cell.getTimestamp());
-
- return pendingGet;
- }
-
- /**
- * Filters the raw results returned from HBase and returns only those belonging to the current snapshot, as defined
- * by the transaction object. If the raw results don't contain enough information for a particular qualifier, it
- * will request more versions from HBase.
- *
- * @param rawCells Raw cells that we are going to filter
- * @param transaction Defines the current snapshot
- * @param versionsToRequest Number of versions requested from hbase
- * @param familyDeletionCache Accumulates the family deletion markers to identify cells that deleted with a higher version
- * @return Filtered KVs belonging to the transaction snapshot
- */
- @Override
- public List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
- int versionsToRequest, Map<String, List<Cell>> familyDeletionCache) throws IOException {
-
- assert (rawCells != null && transaction != null && versionsToRequest >= 1);
-
- List<Cell> keyValuesInSnapshot = new ArrayList<>();
- List<Get> pendingGetsList = new ArrayList<>();
-
- int numberOfVersionsToFetch = versionsToRequest * 2;
- if (numberOfVersionsToFetch < 1) {
- numberOfVersionsToFetch = versionsToRequest;
- }
-
- Map<Long, Long> commitCache = buildCommitCache(rawCells);
- buildFamilyDeletionCache(rawCells, familyDeletionCache);
-
- for (Collection<Cell> columnCells : groupCellsByColumnFilteringShadowCellsAndFamilyDeletion(rawCells)) {
- boolean snapshotValueFound = false;
- Cell oldestCell = null;
- for (Cell cell : columnCells) {
- snapshotValueFound = checkFamilyDeletionCache(cell, transaction, familyDeletionCache, commitCache);
-
- if (snapshotValueFound == true) {
- if (transaction.getVisibilityLevel() == VisibilityLevel.SNAPSHOT_ALL) {
- snapshotValueFound = false;
- } else {
- break;
- }
- }
-
- if (isCellInTransaction(cell, transaction, commitCache) ||
- isCellInSnapshot(cell, transaction, commitCache)) {
- 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 ||
- !isCellInTransaction(cell, transaction, commitCache)) {
- snapshotValueFound = true;
- break;
- }
- }
- oldestCell = cell;
- }
- if (!snapshotValueFound) {
- assert (oldestCell != null);
- Get pendingGet = createPendingGet(oldestCell, numberOfVersionsToFetch);
- pendingGetsList.add(pendingGet);
- }
- }
-
- if (!pendingGetsList.isEmpty()) {
- Result[] pendingGetsResults = tableAccessWrapper.get(pendingGetsList);
- for (Result pendingGetResult : pendingGetsResults) {
- if (!pendingGetResult.isEmpty()) {
- keyValuesInSnapshot.addAll(
- filterCellsForSnapshot(pendingGetResult.listCells(), transaction, numberOfVersionsToFetch, familyDeletionCache));
- }
- }
- }
-
- Collections.sort(keyValuesInSnapshot, KeyValue.COMPARATOR);
-
- return keyValuesInSnapshot;
- }
-
- @Override
- public Result get(TTable ttable, 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, List<Cell>>());
- }
-
- return Result.create(filteredKeyValues);
- }
-
- @Override
- public ResultScanner getScanner(TTable ttable, Scan scan, HBaseTransaction transaction) throws IOException {
-
- return ttable.new TransactionalClientScanner(transaction, scan, 1);
-
- }
-
- @Override
- public boolean isCommitted(HBaseCellId hBaseCellId, long epoch) throws TransactionException {
- try {
- long timestamp = hBaseCellId.getTimestamp() - (hBaseCellId.getTimestamp() % AbstractTransactionManager.MAX_CHECKPOINTS_PER_TXN);
- CommitTimestamp tentativeCommitTimestamp =
- locateCellCommitTimestamp(timestamp, epoch,
- new CommitTimestampLocatorImpl(hBaseCellId, Maps.<Long, Long>newHashMap(), tableAccessWrapper));
-
- // If transaction that added the cell was invalidated
- if (!tentativeCommitTimestamp.isValid()) {
- return false;
- }
-
- switch (tentativeCommitTimestamp.getLocation()) {
- case COMMIT_TABLE:
- case SHADOW_CELL:
- return true;
- case NOT_PRESENT:
- return false;
- case CACHE: // cache was empty
- default:
- return false;
- }
- } catch (IOException e) {
- throw new TransactionException("Failure while checking if a transaction was committed", e);
- }
- }
-
- static ImmutableList<Collection<Cell>> groupCellsByColumnFilteringShadowCellsAndFamilyDeletion(List<Cell> rawCells) {
-
- Predicate<Cell> shadowCellAndFamilyDeletionFilter = new Predicate<Cell>() {
-
- @Override
- public boolean apply(Cell cell) {
- boolean familyDeletionMarkerCondition = CellUtil.matchingQualifier(cell, CellUtils.FAMILY_DELETE_QUALIFIER) &&
- CellUtil.matchingValue(cell, HConstants.EMPTY_BYTE_ARRAY);
-
- return cell != null && !CellUtils.isShadowCell(cell) && !familyDeletionMarkerCondition;
- }
-
- };
-
- 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, shadowCellAndFamilyDeletionFilter), cellToColumnWrapper)
- .asMap().values()
- .asList();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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 12dfb71..9b82148 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
@@ -17,8 +17,12 @@
*/
package org.apache.omid.transaction;
-import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
import com.google.common.base.Optional;
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Multimaps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
@@ -39,16 +43,15 @@ import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.TimeRange;
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.tso.client.OmidClientConfiguration.ConflictDetectionLevel;
+import org.apache.omid.transaction.HBaseTransactionManager.CommitTimestampLocatorImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
@@ -57,7 +60,6 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.NavigableSet;
-import java.util.Set;
/**
* Provides transactional methods for accessing and modifying a given snapshot of data identified by an opaque {@link
@@ -71,8 +73,6 @@ public class TTable implements Closeable {
private HTableInterface table;
- private SnapshotFilter snapshotFilter;
-
// ----------------------------------------------------------------------------------------------------------------
// Construction
// ----------------------------------------------------------------------------------------------------------------
@@ -81,10 +81,6 @@ public class TTable implements Closeable {
this(new HTable(conf, tableName));
}
- public TTable(Configuration conf, byte[] tableName, CommitTable.Client commitTableClient) throws IOException {
- this(new HTable(conf, tableName), commitTableClient);
- }
-
public TTable(String tableName) throws IOException {
this(HBaseConfiguration.create(), Bytes.toBytes(tableName));
}
@@ -93,40 +89,16 @@ public class TTable implements Closeable {
this(conf, Bytes.toBytes(tableName));
}
- public TTable(Configuration conf, String tableName, CommitTable.Client commitTableClient) throws IOException {
- this(conf, Bytes.toBytes(tableName), commitTableClient);
- }
-
public TTable(HTableInterface hTable) throws IOException {
table = hTable;
healerTable = new HTable(table.getConfiguration(), table.getTableName());
- boolean serverSideFilter = table.getConfiguration().getBoolean("omid.server.side.filter", false);
- snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) : new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable));
- }
-
- public TTable(HTableInterface hTable, CommitTable.Client commitTableClient) throws IOException {
- table = hTable;
- healerTable = new HTable(table.getConfiguration(), table.getTableName());
- boolean serverSideFilter = table.getConfiguration().getBoolean("omid.server.side.filter", false);
- snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) : new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable), commitTableClient);
}
public TTable(HTableInterface hTable, HTableInterface healerTable) throws IOException {
table = hTable;
this.healerTable = healerTable;
- Configuration config = table.getConfiguration();
- boolean serverSideFilter = (config == null) ? false : config.getBoolean("omid.server.side.filter", false);
- 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;
- boolean serverSideFilter = table.getConfiguration().getBoolean("omid.server.side.filter", false);
- snapshotFilter = (serverSideFilter) ? new AttributeSetSnapshotFilter(hTable) : new SnapshotFilterImpl(new HTableAccessWrapper(hTable, healerTable), commitTableClient);
- }
-
-
// ----------------------------------------------------------------------------------------------------------------
// Closeable implementation
// ----------------------------------------------------------------------------------------------------------------
@@ -160,7 +132,7 @@ public class TTable implements Closeable {
HBaseTransaction transaction = enforceHBaseTransactionAsParam(tx);
- final long readTimestamp = transaction.getReadTimestamp();
+ final long readTimestamp = transaction.getStartTimestamp();
final Get tsget = new Get(get.getRow()).setFilter(get.getFilter());
TimeRange timeRange = get.getTimeRange();
long startTime = timeRange.getMin();
@@ -177,41 +149,19 @@ public class TTable implements Closeable {
tsget.addColumn(family, qualifier);
tsget.addColumn(family, CellUtils.addShadowCellSuffix(qualifier));
}
- tsget.addColumn(family, CellUtils.FAMILY_DELETE_QUALIFIER);
- tsget.addColumn(family, CellUtils.addShadowCellSuffix(CellUtils.FAMILY_DELETE_QUALIFIER));
}
}
LOG.trace("Initial Get = {}", tsget);
- return snapshotFilter.get(this, tsget, transaction);
- }
-
- private void familyQualifierBasedDeletion(HBaseTransaction tx, Put deleteP, Get deleteG) throws IOException {
- Result result = this.get(tx, deleteG);
+ // Return the KVs that belong to the transaction snapshot, ask for more
+ // versions if needed
+ Result result = table.get(tsget);
+ List<Cell> filteredKeyValues = Collections.emptyList();
if (!result.isEmpty()) {
- for (Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entryF : result.getMap()
- .entrySet()) {
- byte[] family = entryF.getKey();
- for (Entry<byte[], NavigableMap<Long, byte[]>> entryQ : entryF.getValue().entrySet()) {
- byte[] qualifier = entryQ.getKey();
- tx.addWriteSetElement(new HBaseCellId(table, deleteP.getRow(), family, qualifier,
- tx.getWriteTimestamp()));
- }
- deleteP.add(family, CellUtils.FAMILY_DELETE_QUALIFIER, tx.getWriteTimestamp(),
- HConstants.EMPTY_BYTE_ARRAY);
- }
+ filteredKeyValues = filterCellsForSnapshot(result.listCells(), transaction, tsget.getMaxVersions());
}
- }
-
- private void familyQualifierBasedDeletionWithOutRead(HBaseTransaction tx, Put deleteP, Get deleteG) {
- Set<byte[]> fset = deleteG.getFamilyMap().keySet();
- for (byte[] family : fset) {
- deleteP.add(family, CellUtils.FAMILY_DELETE_QUALIFIER, tx.getWriteTimestamp(),
- HConstants.EMPTY_BYTE_ARRAY);
- }
- tx.addWriteSetElement(new HBaseCellId(table, deleteP.getRow(), null, null,
- tx.getWriteTimestamp()));
+ return Result.create(filteredKeyValues);
}
/**
@@ -227,48 +177,47 @@ public class TTable implements Closeable {
HBaseTransaction transaction = enforceHBaseTransactionAsParam(tx);
- final long writeTimestamp = transaction.getStartTimestamp();
- boolean deleteFamily = false;
+ final long startTimestamp = transaction.getStartTimestamp();
+ boolean issueGet = false;
- final Put deleteP = new Put(delete.getRow(), writeTimestamp);
+ final Put deleteP = new Put(delete.getRow(), startTimestamp);
final Get deleteG = new Get(delete.getRow());
Map<byte[], List<Cell>> fmap = delete.getFamilyCellMap();
if (fmap.isEmpty()) {
- familyQualifierBasedDeletion(transaction, deleteP, deleteG);
+ issueGet = true;
}
-
for (List<Cell> cells : fmap.values()) {
for (Cell cell : cells) {
- CellUtils.validateCell(cell, writeTimestamp);
+ CellUtils.validateCell(cell, startTimestamp);
switch (KeyValue.Type.codeToType(cell.getTypeByte())) {
case DeleteColumn:
deleteP.add(CellUtil.cloneFamily(cell),
CellUtil.cloneQualifier(cell),
- writeTimestamp,
+ startTimestamp,
CellUtils.DELETE_TOMBSTONE);
transaction.addWriteSetElement(
new HBaseCellId(table,
delete.getRow(),
CellUtil.cloneFamily(cell),
CellUtil.cloneQualifier(cell),
- writeTimestamp));
+ cell.getTimestamp()));
break;
case DeleteFamily:
deleteG.addFamily(CellUtil.cloneFamily(cell));
- deleteFamily = true;
+ issueGet = true;
break;
case Delete:
if (cell.getTimestamp() == HConstants.LATEST_TIMESTAMP) {
deleteP.add(CellUtil.cloneFamily(cell),
CellUtil.cloneQualifier(cell),
- writeTimestamp,
+ startTimestamp,
CellUtils.DELETE_TOMBSTONE);
transaction.addWriteSetElement(
new HBaseCellId(table,
delete.getRow(),
CellUtil.cloneFamily(cell),
CellUtil.cloneQualifier(cell),
- writeTimestamp));
+ cell.getTimestamp()));
break;
} else {
throw new UnsupportedOperationException(
@@ -279,11 +228,21 @@ public class TTable implements Closeable {
}
}
}
- if (deleteFamily) {
- if (enforceHBaseTransactionManagerAsParam(transaction.getTransactionManager()).getConflictDetectionLevel() == ConflictDetectionLevel.ROW) {
- familyQualifierBasedDeletionWithOutRead(transaction, deleteP, deleteG);
- } else {
- familyQualifierBasedDeletion(transaction, deleteP, deleteG);
+ if (issueGet) {
+ // It's better to perform a transactional get to avoid deleting more
+ // than necessary
+ Result result = this.get(transaction, deleteG);
+ if (!result.isEmpty()) {
+ for (Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entryF : result.getMap()
+ .entrySet()) {
+ byte[] family = entryF.getKey();
+ for (Entry<byte[], NavigableMap<Long, byte[]>> entryQ : entryF.getValue().entrySet()) {
+ byte[] qualifier = entryQ.getKey();
+ deleteP.add(family, qualifier, CellUtils.DELETE_TOMBSTONE);
+ transaction.addWriteSetElement(new HBaseCellId(table, delete.getRow(), family, qualifier,
+ transaction.getStartTimestamp()));
+ }
+ }
}
}
@@ -306,19 +265,18 @@ public class TTable implements Closeable {
HBaseTransaction transaction = enforceHBaseTransactionAsParam(tx);
- final long writeTimestamp = transaction.getWriteTimestamp();
-
+ final long startTimestamp = transaction.getStartTimestamp();
// create put with correct ts
- final Put tsput = new Put(put.getRow(), writeTimestamp);
+ final Put tsput = new Put(put.getRow(), startTimestamp);
Map<byte[], List<Cell>> kvs = put.getFamilyCellMap();
for (List<Cell> kvl : kvs.values()) {
for (Cell c : kvl) {
- CellUtils.validateCell(c, writeTimestamp);
+ CellUtils.validateCell(c, startTimestamp);
// Reach into keyvalue to update timestamp.
// It's not nice to reach into keyvalue internals,
// but we want to avoid having to copy the whole thing
KeyValue kv = KeyValueUtil.ensureKeyValue(c);
- Bytes.putLong(kv.getValueArray(), kv.getTimestampOffset(), writeTimestamp);
+ Bytes.putLong(kv.getValueArray(), kv.getTimestampOffset(), startTimestamp);
tsput.add(kv);
transaction.addWriteSetElement(
@@ -349,7 +307,7 @@ public class TTable implements Closeable {
Scan tsscan = new Scan(scan);
tsscan.setMaxVersions(1);
- tsscan.setTimeRange(0, transaction.getReadTimestamp() + 1);
+ tsscan.setTimeRange(0, transaction.getStartTimestamp() + 1);
Map<byte[], NavigableSet<byte[]>> kvs = scan.getFamilyMap();
for (Map.Entry<byte[], NavigableSet<byte[]>> entry : kvs.entrySet()) {
byte[] family = entry.getKey();
@@ -360,34 +318,181 @@ public class TTable implements Closeable {
for (byte[] qualifier : qualifiers) {
tsscan.addColumn(family, CellUtils.addShadowCellSuffix(qualifier));
}
- if (!qualifiers.isEmpty()) {
- tsscan.addColumn(entry.getKey(), CellUtils.FAMILY_DELETE_QUALIFIER);
+ }
+ return new TransactionalClientScanner(transaction, tsscan, 1);
+ }
+
+ /**
+ * Filters the raw results returned from HBase and returns only those belonging to the current snapshot, as defined
+ * by the transaction object. If the raw results don't contain enough information for a particular qualifier, it
+ * will request more versions from HBase.
+ *
+ * @param rawCells Raw cells that we are going to filter
+ * @param transaction Defines the current snapshot
+ * @param versionsToRequest Number of versions requested from hbase
+ * @return Filtered KVs belonging to the transaction snapshot
+ */
+ List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
+ int versionsToRequest) throws IOException {
+
+ assert (rawCells != null && transaction != null && versionsToRequest >= 1);
+
+ List<Cell> keyValuesInSnapshot = new ArrayList<>();
+ List<Get> pendingGetsList = new ArrayList<>();
+
+ int numberOfVersionsToFetch = versionsToRequest * 2;
+ if (numberOfVersionsToFetch < 1) {
+ numberOfVersionsToFetch = versionsToRequest;
+ }
+
+ Map<Long, Long> commitCache = buildCommitCache(rawCells);
+
+ for (Collection<Cell> columnCells : groupCellsByColumnFilteringShadowCells(rawCells)) {
+ boolean snapshotValueFound = false;
+ Cell oldestCell = null;
+ for (Cell cell : columnCells) {
+ if (isCellInSnapshot(cell, transaction, commitCache)) {
+ if (!CellUtil.matchingValue(cell, CellUtils.DELETE_TOMBSTONE)) {
+ keyValuesInSnapshot.add(cell);
+ }
+ snapshotValueFound = true;
+ break;
+ }
+ oldestCell = cell;
+ }
+ if (!snapshotValueFound) {
+ assert (oldestCell != null);
+ Get pendingGet = createPendingGet(oldestCell, numberOfVersionsToFetch);
+ pendingGetsList.add(pendingGet);
+ }
+ }
+
+ if (!pendingGetsList.isEmpty()) {
+ Result[] pendingGetsResults = table.get(pendingGetsList);
+ for (Result pendingGetResult : pendingGetsResults) {
+ if (!pendingGetResult.isEmpty()) {
+ keyValuesInSnapshot.addAll(
+ filterCellsForSnapshot(pendingGetResult.listCells(), transaction, numberOfVersionsToFetch));
+ }
}
}
- return snapshotFilter.getScanner(this, tsscan, transaction);
+ Collections.sort(keyValuesInSnapshot, KeyValue.COMPARATOR);
+
+ assert (keyValuesInSnapshot.size() <= rawCells.size());
+ return keyValuesInSnapshot;
}
-
- List<Cell> filterCellsForSnapshot(List<Cell> rawCells, HBaseTransaction transaction,
- int versionsToRequest, Map<String, List<Cell>> familyDeletionCache) throws IOException {
- return snapshotFilter.filterCellsForSnapshot(rawCells, transaction, versionsToRequest, familyDeletionCache);
+ private Map<Long, Long> buildCommitCache(List<Cell> rawCells) {
+
+ Map<Long, Long> commitCache = new HashMap<>();
+
+ for (Cell cell : rawCells) {
+ if (CellUtils.isShadowCell(cell)) {
+ commitCache.put(cell.getTimestamp(), Bytes.toLong(CellUtil.cloneValue(cell)));
+ }
+ }
+
+ return commitCache;
}
+ private boolean isCellInSnapshot(Cell kv, HBaseTransaction transaction, Map<Long, Long> commitCache)
+ throws IOException {
+
+ long startTimestamp = transaction.getStartTimestamp();
- public class TransactionalClientScanner implements ResultScanner {
+ if (kv.getTimestamp() == startTimestamp) {
+ return true;
+ }
+
+ Optional<Long> commitTimestamp =
+ tryToLocateCellCommitTimestamp(transaction.getTransactionManager(), transaction.getEpoch(), kv,
+ commitCache);
+
+ return commitTimestamp.isPresent() && commitTimestamp.get() < startTimestamp;
+ }
+
+ private Get createPendingGet(Cell cell, int versionCount) throws IOException {
+
+ Get pendingGet = new Get(CellUtil.cloneRow(cell));
+ pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtil.cloneQualifier(cell));
+ pendingGet.addColumn(CellUtil.cloneFamily(cell), CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
+ cell.getQualifierOffset(),
+ cell.getQualifierLength()));
+ pendingGet.setMaxVersions(versionCount);
+ pendingGet.setTimeRange(0, cell.getTimestamp());
+
+ return pendingGet;
+ }
+
+ private Optional<Long> tryToLocateCellCommitTimestamp(AbstractTransactionManager transactionManager,
+ long epoch,
+ Cell cell,
+ Map<Long, Long> commitCache)
+ throws IOException {
+
+ CommitTimestamp tentativeCommitTimestamp =
+ transactionManager.locateCellCommitTimestamp(
+ cell.getTimestamp(),
+ epoch,
+ new CommitTimestampLocatorImpl(
+ new HBaseCellId(table,
+ CellUtil.cloneRow(cell),
+ CellUtil.cloneFamily(cell),
+ CellUtil.cloneQualifier(cell),
+ cell.getTimestamp()),
+ commitCache));
+
+ // If transaction that added the cell was invalidated
+ if (!tentativeCommitTimestamp.isValid()) {
+ return Optional.absent();
+ }
+
+ switch (tentativeCommitTimestamp.getLocation()) {
+ case COMMIT_TABLE:
+ // If the commit timestamp is found in the persisted commit table,
+ // that means the writing process of the shadow cell in the post
+ // commit phase of the client probably failed, so we heal the shadow
+ // cell with the right commit timestamp for avoiding further reads to
+ // hit the storage
+ healShadowCell(cell, tentativeCommitTimestamp.getValue());
+ return Optional.of(tentativeCommitTimestamp.getValue());
+ case CACHE:
+ case SHADOW_CELL:
+ return Optional.of(tentativeCommitTimestamp.getValue());
+ case NOT_PRESENT:
+ return Optional.absent();
+ default:
+ assert (false);
+ return Optional.absent();
+ }
+ }
+
+ void healShadowCell(Cell cell, long commitTimestamp) {
+ Put put = new Put(CellUtil.cloneRow(cell));
+ byte[] family = CellUtil.cloneFamily(cell);
+ byte[] shadowCellQualifier = CellUtils.addShadowCellSuffix(cell.getQualifierArray(),
+ cell.getQualifierOffset(),
+ cell.getQualifierLength());
+ put.add(family, shadowCellQualifier, cell.getTimestamp(), Bytes.toBytes(commitTimestamp));
+ try {
+ healerTable.put(put);
+ } catch (IOException e) {
+ LOG.warn("Failed healing shadow cell for kv {}", cell, e);
+ }
+ }
+
+ protected class TransactionalClientScanner implements ResultScanner {
private HBaseTransaction state;
private ResultScanner innerScanner;
private int maxVersions;
- Map<String, List<Cell>> familyDeletionCache;
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, List<Cell>>();
}
@@ -400,7 +505,7 @@ public class TTable implements Closeable {
return null;
}
if (!result.isEmpty()) {
- filteredResult = filterCellsForSnapshot(result.listCells(), state, maxVersions, familyDeletionCache);
+ filteredResult = filterCellsForSnapshot(result.listCells(), state, maxVersions);
}
}
return Result.create(filteredResult);
@@ -689,38 +794,28 @@ public class TTable implements Closeable {
}
}
- private HBaseTransactionManager enforceHBaseTransactionManagerAsParam(TransactionManager tm) {
- if (tm instanceof HBaseTransactionManager) {
- return (HBaseTransactionManager) tm;
- } else {
- throw new IllegalArgumentException(
- String.format("The transaction manager object passed %s is not an instance of HBaseTransactionManager ",
- tm.getClass().getName()));
- }
- }
+ static ImmutableList<Collection<Cell>> groupCellsByColumnFilteringShadowCells(List<Cell> rawCells) {
- // For testing
+ Predicate<Cell> shadowCellFilter = new Predicate<Cell>() {
- @VisibleForTesting
- boolean isCommitted(HBaseCellId hBaseCellId, long epoch) throws TransactionException {
- return snapshotFilter.isCommitted(hBaseCellId, epoch);
- }
+ @Override
+ public boolean apply(Cell cell) {
+ return cell != null && !CellUtils.isShadowCell(cell);
+ }
- @VisibleForTesting
- CommitTimestamp locateCellCommitTimestamp(long cellStartTimestamp, long epoch,
- CommitTimestampLocator locator) throws IOException {
- return snapshotFilter.locateCellCommitTimestamp(cellStartTimestamp, epoch, locator);
- }
+ };
- @VisibleForTesting
- Optional<CommitTimestamp> readCommitTimestampFromShadowCell(long cellStartTimestamp, CommitTimestampLocator locator)
- throws IOException
- {
- return snapshotFilter.readCommitTimestampFromShadowCell(cellStartTimestamp, locator);
- }
+ Function<Cell, ColumnWrapper> cellToColumnWrapper = new Function<Cell, ColumnWrapper>() {
- SnapshotFilter getSnapshotFilter() {
- return snapshotFilter;
- }
+ @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();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/31dd269d/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
deleted file mode 100644
index 2fc5362..0000000
--- a/hbase-client/src/main/java/org/apache/omid/transaction/TableAccessWrapper.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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 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 {
-
- public Result[] get(List<Get> get) throws IOException;
- public Result get(Get get) throws IOException;
- public void put(Put put) throws IOException;
-
-}