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/06/03 08:14:47 UTC
incubator-omid git commit: [OMID-100] James Taylor's patch to:
https://issues.apache.org/jira/browse/OMID-100 including OmidCompactor and
OmidSnapshotFilter
Repository: incubator-omid
Updated Branches:
refs/heads/phoenix-integration 5ab09f3fc -> d4dab878d
[OMID-100] James Taylor's patch to:
https://issues.apache.org/jira/browse/OMID-100 including OmidCompactor and
OmidSnapshotFilter
Project: http://git-wip-us.apache.org/repos/asf/incubator-omid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-omid/commit/d4dab878
Tree: http://git-wip-us.apache.org/repos/asf/incubator-omid/tree/d4dab878
Diff: http://git-wip-us.apache.org/repos/asf/incubator-omid/diff/d4dab878
Branch: refs/heads/phoenix-integration
Commit: d4dab878dd749acdc4c34929ae826e609d9983cd
Parents: 5ab09f3
Author: Ohad Shacham <oh...@yahoo-inc.com>
Authored: Sun Jun 3 11:14:30 2018 +0300
Committer: Ohad Shacham <oh...@yahoo-inc.com>
Committed: Sun Jun 3 11:14:30 2018 +0300
----------------------------------------------------------------------
.../hbase/regionserver/OmidRegionScanner.java | 29 ++++++----
.../apache/omid/transaction/OmidCompactor.java | 56 +++++++++++---------
.../omid/transaction/OmidSnapshotFilter.java | 46 ++++++++--------
3 files changed, 75 insertions(+), 56 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/d4dab878/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 6c95f65..a3bcafe 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
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
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;
@@ -111,21 +112,27 @@ public class OmidRegionScanner implements RegionScanner {
}
public boolean nextRaw(List<Cell> result, int limit) throws IOException {
- List<Cell> filteredResult = new ArrayList<Cell>();
- while (filteredResult.isEmpty()) {
- scanner.nextRaw(filteredResult);
- if (filteredResult.isEmpty()) {
- return false;
+ 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);
}
- filteredResult = snapshotFilter.filterCellsForSnapshot(filteredResult, transaction, maxVersions, familyDeletionCache, attributeMap);
- }
+ for (Cell cell : filteredResult) {
+ result.add(cell);
+ }
- for (Cell cell : filteredResult) {
- result.add(cell);
+ return true;
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new DoNotRetryIOException(e);
}
-
- return true;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/d4dab878/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidCompactor.java
----------------------------------------------------------------------
diff --git a/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidCompactor.java b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidCompactor.java
index e49f4bf..e839187 100644
--- a/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidCompactor.java
+++ b/hbase-coprocessor/src/main/java/org/apache/omid/transaction/OmidCompactor.java
@@ -18,11 +18,13 @@
package org.apache.omid.transaction;
import com.google.common.annotations.VisibleForTesting;
+
import org.apache.omid.committable.CommitTable;
import org.apache.omid.committable.hbase.HBaseCommitTable;
import org.apache.omid.committable.hbase.HBaseCommitTableConfig;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
@@ -108,37 +110,43 @@ public class OmidCompactor extends BaseRegionObserver {
}
@Override
- public InternalScanner preCompact(ObserverContext<RegionCoprocessorEnvironment> e,
+ public InternalScanner preCompact(ObserverContext<RegionCoprocessorEnvironment> env,
Store store,
InternalScanner scanner,
ScanType scanType,
CompactionRequest request) throws IOException {
boolean omidCompactable;
- if (enableCompactorForAllFamilies) {
- omidCompactable = true;
- } else {
- HTableDescriptor desc = e.getEnvironment().getRegion().getTableDesc();
- HColumnDescriptor famDesc
- = desc.getFamily(Bytes.toBytes(store.getColumnFamilyName()));
- omidCompactable = Boolean.valueOf(famDesc.getValue(OMID_COMPACTABLE_CF_FLAG));
- }
+ try {
+ if (enableCompactorForAllFamilies) {
+ omidCompactable = true;
+ } else {
+ HTableDescriptor desc = env.getEnvironment().getRegion().getTableDesc();
+ HColumnDescriptor famDesc
+ = desc.getFamily(Bytes.toBytes(store.getColumnFamilyName()));
+ omidCompactable = Boolean.valueOf(famDesc.getValue(OMID_COMPACTABLE_CF_FLAG));
+ }
- // only column families tagged as compactable are compacted
- // with omid compactor
- if (!omidCompactable) {
- return scanner;
- } else {
- CommitTable.Client commitTableClient = commitTableClientQueue.poll();
- if (commitTableClient == null) {
- commitTableClient = initAndGetCommitTableClient();
+ // only column families tagged as compactable are compacted
+ // with omid compactor
+ if (!omidCompactable) {
+ return scanner;
+ } else {
+ CommitTable.Client commitTableClient = commitTableClientQueue.poll();
+ if (commitTableClient == null) {
+ commitTableClient = initAndGetCommitTableClient();
+ }
+ boolean isMajorCompaction = request.isMajor();
+ return new CompactorScanner(env,
+ scanner,
+ commitTableClient,
+ commitTableClientQueue,
+ isMajorCompaction,
+ retainNonTransactionallyDeletedCells);
}
- boolean isMajorCompaction = request.isMajor();
- return new CompactorScanner(e,
- scanner,
- commitTableClient,
- commitTableClientQueue,
- isMajorCompaction,
- retainNonTransactionallyDeletedCells);
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new DoNotRetryIOException(e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-omid/blob/d4dab878/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 6c2327a..2cd7efb 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
@@ -25,10 +25,10 @@ import org.apache.omid.committable.hbase.HBaseCommitTableConfig;
import org.apache.omid.proto.TSOProto;
import org.apache.omid.transaction.AbstractTransaction.VisibilityLevel;
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;
@@ -38,7 +38,6 @@ 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;
@@ -109,32 +108,37 @@ public class OmidSnapshotFilter extends BaseRegionObserver {
if (get.getAttribute(CellUtils.CLIENT_GET_ATTRIBUTE) == null) return;
- 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
+ try {
+ get.setAttribute(CellUtils.CLIENT_GET_ATTRIBUTE, null);
+ RegionAccessWrapper regionAccessWrapper = new RegionAccessWrapper(HBaseShims.getRegionCoprocessorRegion(c.getEnvironment()));
+ Result res = regionAccessWrapper.get(get); // get parameters were set at the client side
- snapshotFilter.setTableAccessWrapper(regionAccessWrapper);
+ snapshotFilter.setTableAccessWrapper(regionAccessWrapper);
- List<Cell> filteredKeyValues = Collections.emptyList();
- if (!res.isEmpty()) {
- TSOProto.Transaction transaction = TSOProto.Transaction.parseFrom(get.getAttribute(CellUtils.TRANSACTION_ATTRIBUTE));
+ List<Cell> filteredKeyValues = Collections.emptyList();
+ if (!res.isEmpty()) {
+ TSOProto.Transaction transaction = TSOProto.Transaction.parseFrom(get.getAttribute(CellUtils.TRANSACTION_ATTRIBUTE));
- long id = transaction.getTimestamp();
- long readTs = transaction.getReadTimestamp();
- long epoch = transaction.getEpoch();
- VisibilityLevel visibilityLevel = VisibilityLevel.fromInteger(transaction.getVisibilityLevel());
+ 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);
- filteredKeyValues = snapshotFilter.filterCellsForSnapshot(res.listCells(), hbaseTransaction, get.getMaxVersions(), new HashMap<String, List<Cell>>(), get.getAttributesMap());
- }
+ 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, List<Cell>>(), get.getAttributesMap());
+ }
- for (Cell cell : filteredKeyValues) {
- result.add(cell);
- }
-
- c.bypass();
+ for (Cell cell : filteredKeyValues) {
+ result.add(cell);
+ }
+ c.bypass();
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new DoNotRetryIOException(e);
+ }
}
@Override