You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by jm...@apache.org on 2013/02/13 21:58:32 UTC
svn commit: r1445918 [7/29] - in /hbase/branches/hbase-7290: ./ bin/ conf/
dev-support/ hbase-client/ hbase-common/
hbase-common/src/main/java/org/apache/hadoop/hbase/
hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/
hbase-common/src/mai...
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Increment.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Increment.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Increment.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Increment.java Wed Feb 13 20:58:23 2013
@@ -45,7 +45,6 @@ import org.apache.hadoop.hbase.util.Byte
@InterfaceStability.Stable
public class Increment implements Row {
private byte [] row = null;
- private long lockId = -1L;
private boolean writeToWAL = true;
private TimeRange tr = new TimeRange();
private Map<byte [], NavigableMap<byte [], Long>> familyMap =
@@ -55,31 +54,17 @@ public class Increment implements Row {
public Increment() {}
/**
- * Create a Increment operation for the specified row.
- * <p>
- * At least one column must be incremented.
- * @param row row key
- */
- public Increment(byte [] row) {
- this(row, null);
- }
-
- /**
* Create a Increment operation for the specified row, using an existing row
* lock.
* <p>
* At least one column must be incremented.
* @param row row key
- * @param rowLock previously acquired row lock, or null
*/
- public Increment(byte [] row, RowLock rowLock) {
+ public Increment(byte [] row) {
if (row == null) {
throw new IllegalArgumentException("Cannot increment a null row");
}
this.row = row;
- if(rowLock != null) {
- this.lockId = rowLock.getLockId();
- }
}
/**
@@ -119,22 +104,6 @@ public class Increment implements Row {
}
/**
- * Method for retrieving the increment's RowLock
- * @return RowLock
- */
- public RowLock getRowLock() {
- return new RowLock(this.row, this.lockId);
- }
-
- /**
- * Method for retrieving the increment's lockId
- * @return lockId
- */
- public long getLockId() {
- return this.lockId;
- }
-
- /**
* Method for retrieving whether WAL will be written to or not
* @return true if WAL should be used, false if not
*/
@@ -274,4 +243,16 @@ public class Increment implements Row {
public int compareTo(Row i) {
return Bytes.compareTo(this.getRow(), i.getRow());
}
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ Row other = (Row) obj;
+ return compareTo(other) == 0;
+ }
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Mutation.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Mutation.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Mutation.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Mutation.java Wed Feb 13 20:58:23 2013
@@ -40,7 +40,6 @@ public abstract class Mutation extends O
protected byte [] row = null;
protected long ts = HConstants.LATEST_TIMESTAMP;
- protected long lockId = -1L;
protected boolean writeToWAL = true;
protected Map<byte [], List<KeyValue>> familyMap =
new TreeMap<byte [], List<KeyValue>>(Bytes.BYTES_COMPARATOR);
@@ -165,23 +164,6 @@ public abstract class Mutation extends O
}
/**
- * Method for retrieving the delete's RowLock
- * @return RowLock
- */
- public RowLock getRowLock() {
- return new RowLock(this.row, this.lockId);
- }
-
- /**
- * Method for retrieving the delete's lock ID.
- *
- * @return The lock ID.
- */
- public long getLockId() {
- return this.lockId;
- }
-
- /**
* Method for retrieving the timestamp
* @return timestamp
*/
@@ -194,6 +176,7 @@ public abstract class Mutation extends O
* @param clusterId
*/
public void setClusterId(UUID clusterId) {
+ if (clusterId == null) return;
byte[] val = new byte[2*Bytes.SIZEOF_LONG];
Bytes.putLong(val, 0, clusterId.getMostSignificantBits());
Bytes.putLong(val, Bytes.SIZEOF_LONG, clusterId.getLeastSignificantBits());
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/OperationWithAttributes.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/OperationWithAttributes.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/OperationWithAttributes.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/OperationWithAttributes.java Wed Feb 13 20:58:23 2013
@@ -35,7 +35,7 @@ public abstract class OperationWithAttri
private Map<String, byte[]> attributes;
// used for uniquely identifying an operation
- static public String ID_ATRIBUTE = "_operation.attributes.id";
+ public static final String ID_ATRIBUTE = "_operation.attributes.id";
public void setAttribute(String name, byte[] value) {
if (attributes == null && value == null) {
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Put.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Put.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Put.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Put.java Wed Feb 13 20:58:23 2013
@@ -46,7 +46,7 @@ import java.util.TreeMap;
public class Put extends Mutation implements HeapSize, Comparable<Row> {
private static final long OVERHEAD = ClassSize.align(
ClassSize.OBJECT + 2 * ClassSize.REFERENCE +
- 2 * Bytes.SIZEOF_LONG + Bytes.SIZEOF_BOOLEAN +
+ 1 * Bytes.SIZEOF_LONG + Bytes.SIZEOF_BOOLEAN +
ClassSize.REFERENCE + ClassSize.TREEMAP);
/**
@@ -54,16 +54,7 @@ public class Put extends Mutation implem
* @param row row key
*/
public Put(byte [] row) {
- this(row, null);
- }
-
- /**
- * Create a Put operation for the specified row, using an existing row lock.
- * @param row row key
- * @param rowLock previously acquired row lock, or null
- */
- public Put(byte [] row, RowLock rowLock) {
- this(row, HConstants.LATEST_TIMESTAMP, rowLock);
+ this(row, HConstants.LATEST_TIMESTAMP);
}
/**
@@ -73,24 +64,11 @@ public class Put extends Mutation implem
* @param ts timestamp
*/
public Put(byte[] row, long ts) {
- this(row, ts, null);
- }
-
- /**
- * Create a Put operation for the specified row, using a given timestamp, and an existing row lock.
- * @param row row key
- * @param ts timestamp
- * @param rowLock previously acquired row lock, or null
- */
- public Put(byte [] row, long ts, RowLock rowLock) {
if(row == null || row.length > HConstants.MAX_ROW_LENGTH) {
throw new IllegalArgumentException("Row key is invalid");
}
this.row = Arrays.copyOf(row, row.length);
this.ts = ts;
- if(rowLock != null) {
- this.lockId = rowLock.getLockId();
- }
}
/**
@@ -98,7 +76,7 @@ public class Put extends Mutation implem
* @param putToCopy put to copy
*/
public Put(Put putToCopy) {
- this(putToCopy.getRow(), putToCopy.ts, putToCopy.getRowLock());
+ this(putToCopy.getRow(), putToCopy.ts);
this.familyMap =
new TreeMap<byte [], List<KeyValue>>(Bytes.BYTES_COMPARATOR);
for(Map.Entry<byte [], List<KeyValue>> entry :
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedException.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedException.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedException.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedException.java Wed Feb 13 20:58:23 2013
@@ -77,8 +77,10 @@ public class RetriesExhaustedException e
* @param exceptions List of exceptions that failed before giving up
*/
public RetriesExhaustedException(final int numTries,
- final List<ThrowableWithExtraContext> exceptions) {
- super(getMessage(numTries, exceptions));
+ final List<ThrowableWithExtraContext> exceptions) {
+ super(getMessage(numTries, exceptions),
+ (exceptions != null && !exceptions.isEmpty() ?
+ exceptions.get(exceptions.size() - 1).t : null));
}
private static String getMessage(String callableVitals, int numTries,
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedWithDetailsException.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedWithDetailsException.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedWithDetailsException.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/RetriesExhaustedWithDetailsException.java Wed Feb 13 20:58:23 2013
@@ -22,7 +22,10 @@ package org.apache.hadoop.hbase.client;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.util.Bytes;
+import java.io.PrintWriter;
+import java.io.StringWriter;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
@@ -54,7 +57,7 @@ extends RetriesExhaustedException {
super("Failed " + exceptions.size() + " action" +
pluralize(exceptions) + ": " +
getDesc(exceptions, actions, hostnameAndPort));
-
+
this.exceptions = exceptions;
this.actions = actions;
this.hostnameAndPort = hostnameAndPort;
@@ -116,6 +119,24 @@ extends RetriesExhaustedException {
return s;
}
+ public String getExhaustiveDescription() {
+ StringWriter errorWriter = new StringWriter();
+ for (int i = 0; i < this.exceptions.size(); ++i) {
+ Throwable t = this.exceptions.get(i);
+ Row action = this.actions.get(i);
+ String server = this.hostnameAndPort.get(i);
+ errorWriter.append("Error #" + i + " from [" + server + "] for ["
+ + ((action == null) ? "unknown key" : Bytes.toStringBinary(action.getRow())) + "]");
+ if (t != null) {
+ PrintWriter pw = new PrintWriter(errorWriter);
+ t.printStackTrace(pw);
+ pw.flush();
+ }
+ }
+ return errorWriter.toString();
+ }
+
+
public static Map<String, Integer> classifyExs(List<Throwable> ths) {
Map<String, Integer> cls = new HashMap<String, Integer>();
for (Throwable t : ths) {
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Scan.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Scan.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Scan.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/Scan.java Wed Feb 13 20:58:23 2013
@@ -92,12 +92,13 @@ public class Scan extends OperationWithA
private int storeLimit = -1;
private int storeOffset = 0;
-
+ private boolean getScan;
+
// If application wants to collect scan metrics, it needs to
// call scan.setAttribute(SCAN_ATTRIBUTES_ENABLE, Bytes.toBytes(Boolean.TRUE))
- static public String SCAN_ATTRIBUTES_METRICS_ENABLE =
+ static public final String SCAN_ATTRIBUTES_METRICS_ENABLE =
"scan.attributes.metrics.enable";
- static public String SCAN_ATTRIBUTES_METRICS_DATA =
+ static public final String SCAN_ATTRIBUTES_METRICS_DATA =
"scan.attributes.metrics.data";
/*
@@ -110,6 +111,7 @@ public class Scan extends OperationWithA
private TimeRange tr = new TimeRange();
private Map<byte [], NavigableSet<byte []>> familyMap =
new TreeMap<byte [], NavigableSet<byte []>>(Bytes.BYTES_COMPARATOR);
+ private Boolean loadColumnFamiliesOnDemand = null;
/**
* Create a Scan operation across all rows.
@@ -140,6 +142,8 @@ public class Scan extends OperationWithA
public Scan(byte [] startRow, byte [] stopRow) {
this.startRow = startRow;
this.stopRow = stopRow;
+ //if the startRow and stopRow both are empty, it is not a Get
+ this.getScan = isStartRowAndEqualsStopRow();
}
/**
@@ -158,7 +162,9 @@ public class Scan extends OperationWithA
caching = scan.getCaching();
maxResultSize = scan.getMaxResultSize();
cacheBlocks = scan.getCacheBlocks();
+ getScan = scan.isGetScan();
filter = scan.getFilter(); // clone?
+ loadColumnFamiliesOnDemand = scan.getLoadColumnFamiliesOnDemandValue();
TimeRange ctr = scan.getTimeRange();
tr = new TimeRange(ctr.getMin(), ctr.getMax());
Map<byte[], NavigableSet<byte[]>> fams = scan.getFamilyMap();
@@ -192,13 +198,17 @@ public class Scan extends OperationWithA
this.storeOffset = get.getRowOffsetPerColumnFamily();
this.tr = get.getTimeRange();
this.familyMap = get.getFamilyMap();
+ this.getScan = true;
}
public boolean isGetScan() {
- return this.startRow != null && this.startRow.length > 0 &&
- Bytes.equals(this.startRow, this.stopRow);
+ return this.getScan || isStartRowAndEqualsStopRow();
}
+ private boolean isStartRowAndEqualsStopRow() {
+ return this.startRow != null && this.startRow.length > 0 &&
+ Bytes.equals(this.startRow, this.stopRow);
+ }
/**
* Get all columns from the specified family.
* <p>
@@ -519,6 +529,41 @@ public class Scan extends OperationWithA
}
/**
+ * Set the value indicating whether loading CFs on demand should be allowed (cluster
+ * default is false). On-demand CF loading doesn't load column families until necessary, e.g.
+ * if you filter on one column, the other column family data will be loaded only for the rows
+ * that are included in result, not all rows like in normal case.
+ * With column-specific filters, like SingleColumnValueFilter w/filterIfMissing == true,
+ * this can deliver huge perf gains when there's a cf with lots of data; however, it can
+ * also lead to some inconsistent results, as follows:
+ * - if someone does a concurrent update to both column families in question you may get a row
+ * that never existed, e.g. for { rowKey = 5, { cat_videos => 1 }, { video => "my cat" } }
+ * someone puts rowKey 5 with { cat_videos => 0 }, { video => "my dog" }, concurrent scan
+ * filtering on "cat_videos == 1" can get { rowKey = 5, { cat_videos => 1 },
+ * { video => "my dog" } }.
+ * - if there's a concurrent split and you have more than 2 column families, some rows may be
+ * missing some column families.
+ */
+ public void setLoadColumnFamiliesOnDemand(boolean value) {
+ this.loadColumnFamiliesOnDemand = value;
+ }
+
+ /**
+ * Get the raw loadColumnFamiliesOnDemand setting; if it's not set, can be null.
+ */
+ public Boolean getLoadColumnFamiliesOnDemandValue() {
+ return this.loadColumnFamiliesOnDemand;
+ }
+
+ /**
+ * Get the logical value indicating whether on-demand CF loading should be allowed.
+ */
+ public boolean doLoadColumnFamiliesOnDemand() {
+ return (this.loadColumnFamiliesOnDemand != null)
+ && this.loadColumnFamiliesOnDemand.booleanValue();
+ }
+
+ /**
* Compile the table and column family (i.e. schema) information
* into a String. Useful for parsing and aggregation by debugging,
* logging, and administration tools.
@@ -547,7 +592,7 @@ public class Scan extends OperationWithA
* Useful for debugging, logging, and administration tools.
* @param maxCols a limit on the number of columns output prior to truncation
* @return Map
- */
+ */
@Override
public Map<String, Object> toMap(int maxCols) {
// start with the fingerpring map and build on top of it
@@ -564,6 +609,7 @@ public class Scan extends OperationWithA
map.put("caching", this.caching);
map.put("maxResultSize", this.maxResultSize);
map.put("cacheBlocks", this.cacheBlocks);
+ map.put("loadColumnFamiliesOnDemand", this.loadColumnFamiliesOnDemand);
List<Long> timeRange = new ArrayList<Long>();
timeRange.add(this.tr.getMin());
timeRange.add(this.tr.getMax());
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java Wed Feb 13 20:58:23 2013
@@ -140,8 +140,9 @@ public class ScannerCallable extends Ser
incRPCcallsMetrics();
ScanRequest request =
RequestConverter.buildScanRequest(scannerId, caching, false, nextCallSeq);
+ ScanResponse response = null;
try {
- ScanResponse response = server.scan(null, request);
+ response = server.scan(null, request);
// Client and RS maintain a nextCallSeq number during the scan. Every next() call
// from client to server will increment this number in both sides. Client passes this
// number along with the request and at RS side both the incoming nextCallSeq and its
@@ -171,7 +172,7 @@ public class ScannerCallable extends Ser
} catch (ServiceException se) {
throw ProtobufUtil.getRemoteException(se);
}
- updateResultsMetrics(rrs);
+ updateResultsMetrics(response);
} catch (IOException e) {
if (logScannerActivity) {
LOG.info("Got exception in fetching from scanner="
@@ -226,22 +227,15 @@ public class ScannerCallable extends Ser
}
}
- private void updateResultsMetrics(Result[] rrs) {
- if (this.scanMetrics == null || rrs == null) {
+ private void updateResultsMetrics(ScanResponse response) {
+ if (this.scanMetrics == null || !response.hasResultSizeBytes()) {
return;
}
- /*
- * broken by protobufs
- for (Result rr : rrs) {
- if (rr.getBytes() != null) {
- this.scanMetrics.countOfBytesInResults.inc(rr.getBytes().getLength());
- if (isRegionServerRemote) {
- this.scanMetrics.countOfBytesInRemoteResults.inc(
- rr.getBytes().getLength());
- }
- }
+ long value = response.getResultSizeBytes();
+ this.scanMetrics.countOfBytesInResults.addAndGet(value);
+ if (isRegionServerRemote) {
+ this.scanMetrics.countOfBytesInRemoteResults.addAndGet(value);
}
- */
}
private void close() {
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ServerCallable.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ServerCallable.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ServerCallable.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ServerCallable.java Wed Feb 13 20:58:23 2013
@@ -33,8 +33,7 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
-import org.apache.hadoop.hbase.client.ClientProtocol;
-import org.apache.hadoop.hbase.ipc.HBaseRPC;
+import org.apache.hadoop.hbase.ipc.HBaseClientRPC;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.ipc.RemoteException;
@@ -113,12 +112,12 @@ public abstract class ServerCallable<T>
}
public void beforeCall() {
- HBaseRPC.setRpcTimeout(this.callTimeout);
+ HBaseClientRPC.setRpcTimeout(this.callTimeout);
this.startTime = System.currentTimeMillis();
}
public void afterCall() {
- HBaseRPC.resetRpcTimeout();
+ HBaseClientRPC.resetRpcTimeout();
this.endTime = System.currentTimeMillis();
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/AggregationClient.java Wed Feb 13 20:58:23 2013
@@ -20,6 +20,10 @@
package org.apache.hadoop.hbase.client.coprocessor;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -51,6 +55,7 @@ import org.apache.hadoop.hbase.util.Byte
import org.apache.hadoop.hbase.util.Pair;
import com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
/**
* This client class is for invoking the aggregate functions deployed on the
@@ -98,7 +103,8 @@ public class AggregationClient {
* The caller is supposed to handle the exception as they are thrown
* & propagated to it.
*/
- public <R, S> R max(final byte[] tableName, final ColumnInterpreter<R, S> ci,
+ public <R, S, P extends Message, Q extends Message, T extends Message>
+ R max(final byte[] tableName, final ColumnInterpreter<R, S, P, Q, T> ci,
final Scan scan) throws Throwable {
final AggregateArgument requestArg = validateArgAndGetPB(scan, ci);
class MaxCallBack implements Batch.Callback<R> {
@@ -130,9 +136,9 @@ public class AggregationClient {
throw controller.getFailedOn();
}
if (response.getFirstPartCount() > 0) {
- return ci.castToCellType(
- ci.parseResponseAsPromotedType(
- getBytesFromResponse(response.getFirstPart(0))));
+ ByteString b = response.getFirstPart(0);
+ Q q = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 3, b);
+ return ci.getCellValueFromProto(q);
}
return null;
}
@@ -168,7 +174,8 @@ public class AggregationClient {
* @return min val <R>
* @throws Throwable
*/
- public <R, S> R min(final byte[] tableName, final ColumnInterpreter<R, S> ci,
+ public <R, S, P extends Message, Q extends Message, T extends Message>
+ R min(final byte[] tableName, final ColumnInterpreter<R, S, P, Q, T> ci,
final Scan scan) throws Throwable {
final AggregateArgument requestArg = validateArgAndGetPB(scan, ci);
class MinCallBack implements Batch.Callback<R> {
@@ -202,9 +209,9 @@ public class AggregationClient {
throw controller.getFailedOn();
}
if (response.getFirstPartCount() > 0) {
- return ci.castToCellType(
- ci.parseResponseAsPromotedType(
- getBytesFromResponse(response.getFirstPart(0))));
+ ByteString b = response.getFirstPart(0);
+ Q q = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 3, b);
+ return ci.getCellValueFromProto(q);
}
return null;
}
@@ -231,8 +238,9 @@ public class AggregationClient {
* @return <R, S>
* @throws Throwable
*/
- public <R, S> long rowCount(final byte[] tableName,
- final ColumnInterpreter<R, S> ci, final Scan scan) throws Throwable {
+ public <R, S, P extends Message, Q extends Message, T extends Message>
+ long rowCount(final byte[] tableName,
+ final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
final AggregateArgument requestArg = validateArgAndGetPB(scan, ci);
class RowNumCallback implements Batch.Callback<Long> {
private final AtomicLong rowCountL = new AtomicLong(0);
@@ -285,7 +293,8 @@ public class AggregationClient {
* @return sum <S>
* @throws Throwable
*/
- public <R, S> S sum(final byte[] tableName, final ColumnInterpreter<R, S> ci,
+ public <R, S, P extends Message, Q extends Message, T extends Message>
+ S sum(final byte[] tableName, final ColumnInterpreter<R, S, P, Q, T> ci,
final Scan scan) throws Throwable {
final AggregateArgument requestArg = validateArgAndGetPB(scan, ci);
@@ -320,8 +329,10 @@ public class AggregationClient {
if (response.getFirstPartCount() == 0) {
return null;
}
- return ci.parseResponseAsPromotedType(
- getBytesFromResponse(response.getFirstPart(0)));
+ ByteString b = response.getFirstPart(0);
+ T t = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 4, b);
+ S s = ci.getPromotedValueFromProto(t);
+ return s;
}
}, sumCallBack);
} finally {
@@ -340,8 +351,9 @@ public class AggregationClient {
* @param scan
* @throws Throwable
*/
- private <R, S> Pair<S, Long> getAvgArgs(final byte[] tableName,
- final ColumnInterpreter<R, S> ci, final Scan scan) throws Throwable {
+ private <R, S, P extends Message, Q extends Message, T extends Message>
+ Pair<S, Long> getAvgArgs(final byte[] tableName,
+ final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
final AggregateArgument requestArg = validateArgAndGetPB(scan, ci);
class AvgCallBack implements Batch.Callback<Pair<S, Long>> {
S sum = null;
@@ -379,8 +391,10 @@ public class AggregationClient {
if (response.getFirstPartCount() == 0) {
return pair;
}
- pair.setFirst(ci.parseResponseAsPromotedType(
- getBytesFromResponse(response.getFirstPart(0))));
+ ByteString b = response.getFirstPart(0);
+ T t = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 4, b);
+ S s = ci.getPromotedValueFromProto(t);
+ pair.setFirst(s);
ByteBuffer bb = ByteBuffer.allocate(8).put(
getBytesFromResponse(response.getSecondPart()));
bb.rewind();
@@ -408,8 +422,9 @@ public class AggregationClient {
* @return <R, S>
* @throws Throwable
*/
- public <R, S> double avg(final byte[] tableName,
- final ColumnInterpreter<R, S> ci, Scan scan) throws Throwable {
+ public <R, S, P extends Message, Q extends Message, T extends Message>
+ double avg(final byte[] tableName,
+ final ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) throws Throwable {
Pair<S, Long> p = getAvgArgs(tableName, ci, scan);
return ci.divideForAvg(p.getFirst(), p.getSecond());
}
@@ -425,8 +440,9 @@ public class AggregationClient {
* @return
* @throws Throwable
*/
- private <R, S> Pair<List<S>, Long> getStdArgs(final byte[] tableName,
- final ColumnInterpreter<R, S> ci, final Scan scan) throws Throwable {
+ private <R, S, P extends Message, Q extends Message, T extends Message>
+ Pair<List<S>, Long> getStdArgs(final byte[] tableName,
+ final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
final AggregateArgument requestArg = validateArgAndGetPB(scan, ci);
class StdCallback implements Batch.Callback<Pair<List<S>, Long>> {
long rowCountVal = 0l;
@@ -474,8 +490,10 @@ public class AggregationClient {
}
List<S> list = new ArrayList<S>();
for (int i = 0; i < response.getFirstPartCount(); i++) {
- list.add(ci.parseResponseAsPromotedType(
- getBytesFromResponse(response.getFirstPart(i))));
+ ByteString b = response.getFirstPart(i);
+ T t = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 4, b);
+ S s = ci.getPromotedValueFromProto(t);
+ list.add(s);
}
pair.setFirst(list);
ByteBuffer bb = ByteBuffer.allocate(8).put(
@@ -505,7 +523,8 @@ public class AggregationClient {
* @return <R, S>
* @throws Throwable
*/
- public <R, S> double std(final byte[] tableName, ColumnInterpreter<R, S> ci,
+ public <R, S, P extends Message, Q extends Message, T extends Message>
+ double std(final byte[] tableName, ColumnInterpreter<R, S, P, Q, T> ci,
Scan scan) throws Throwable {
Pair<List<S>, Long> p = getStdArgs(tableName, ci, scan);
double res = 0d;
@@ -528,9 +547,10 @@ public class AggregationClient {
* (sum of values, sum of weights) for all the regions chosen
* @throws Throwable
*/
- private <R, S> Pair<NavigableMap<byte[], List<S>>, List<S>>
+ private <R, S, P extends Message, Q extends Message, T extends Message>
+ Pair<NavigableMap<byte[], List<S>>, List<S>>
getMedianArgs(final byte[] tableName,
- final ColumnInterpreter<R, S> ci, final Scan scan) throws Throwable {
+ final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
final AggregateArgument requestArg = validateArgAndGetPB(scan, ci);
final NavigableMap<byte[], List<S>> map =
new TreeMap<byte[], List<S>>(Bytes.BYTES_COMPARATOR);
@@ -572,8 +592,10 @@ public class AggregationClient {
List<S> list = new ArrayList<S>();
for (int i = 0; i < response.getFirstPartCount(); i++) {
- list.add(ci.parseResponseAsPromotedType(
- getBytesFromResponse(response.getFirstPart(i))));
+ ByteString b = response.getFirstPart(i);
+ T t = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 4, b);
+ S s = ci.getPromotedValueFromProto(t);
+ list.add(s);
}
return list;
}
@@ -597,7 +619,8 @@ public class AggregationClient {
* @return R the median
* @throws Throwable
*/
- public <R, S> R median(final byte[] tableName, ColumnInterpreter<R, S> ci,
+ public <R, S, P extends Message, Q extends Message, T extends Message>
+ R median(final byte[] tableName, ColumnInterpreter<R, S, P, Q, T> ci,
Scan scan) throws Throwable {
Pair<NavigableMap<byte[], List<S>>, List<S>> p = getMedianArgs(tableName, ci, scan);
byte[] startRow = null;
@@ -672,16 +695,17 @@ public class AggregationClient {
return null;
}
- <R,S>AggregateArgument validateArgAndGetPB(Scan scan, ColumnInterpreter<R,S> ci)
+ <R, S, P extends Message, Q extends Message, T extends Message> AggregateArgument
+ validateArgAndGetPB(Scan scan, ColumnInterpreter<R,S,P,Q,T> ci)
throws IOException {
validateParameters(scan);
final AggregateArgument.Builder requestBuilder =
AggregateArgument.newBuilder();
requestBuilder.setInterpreterClassName(ci.getClass().getCanonicalName());
- ByteString columnInterpreterSpecificData = null;
- if ((columnInterpreterSpecificData = ci.columnInterpreterSpecificData())
+ P columnInterpreterSpecificData = null;
+ if ((columnInterpreterSpecificData = ci.getRequestData())
!= null) {
- requestBuilder.setInterpreterSpecificBytes(columnInterpreterSpecificData);
+ requestBuilder.setInterpreterSpecificBytes(columnInterpreterSpecificData.toByteString());
}
requestBuilder.setScan(ProtobufUtil.toScan(scan));
return requestBuilder.build();
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/Batch.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/Batch.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/Batch.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/Batch.java Wed Feb 13 20:58:23 2013
@@ -19,18 +19,10 @@
package org.apache.hadoop.hbase.client.coprocessor;
-import org.apache.commons.lang.reflect.MethodUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import java.io.IOException;
+
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
-
-import java.io.IOException;
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.lang.reflect.Proxy;
/**
@@ -40,109 +32,20 @@ import java.lang.reflect.Proxy;
@InterfaceAudience.Public
@InterfaceStability.Evolving
public abstract class Batch {
- private static Log LOG = LogFactory.getLog(Batch.class);
-
- /**
- * Creates a new {@link Batch.Call} instance that invokes a method
- * with the given parameters and returns the result.
- *
- * <p>
- * Note that currently the method is naively looked up using the method name
- * and class types of the passed arguments, which means that
- * <em>none of the arguments can be <code>null</code></em>.
- * For more flexibility, see
- * {@link Batch#forMethod(java.lang.reflect.Method, Object...)}.
- * </p>
- *
- * @param protocol the protocol class being called
- * @param method the method name
- * @param args zero or more arguments to be passed to the method
- * (individual args cannot be <code>null</code>!)
- * @param <T> the class type of the protocol implementation being invoked
- * @param <R> the return type for the method call
- * @return a {@code Callable} instance that will invoke the given method
- * and return the results
- * @throws NoSuchMethodException if the method named, with the given argument
- * types, cannot be found in the protocol class
- * @see Batch#forMethod(java.lang.reflect.Method, Object...)
- * @see org.apache.hadoop.hbase.client.HTable#coprocessorExec(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback)
- */
- @Deprecated
- public static <T extends CoprocessorProtocol,R> Call<T,R> forMethod(
- final Class<T> protocol, final String method, final Object... args)
- throws NoSuchMethodException {
- Class[] types = new Class[args.length];
- for (int i=0; i<args.length; i++) {
- if (args[i] == null) {
- throw new NullPointerException("Method argument cannot be null");
- }
- types[i] = args[i].getClass();
- }
-
- Method m = MethodUtils.getMatchingAccessibleMethod(protocol, method, types);
- if (m == null) {
- throw new NoSuchMethodException("No matching method found for '" +
- method + "'");
- }
-
- m.setAccessible(true);
- return forMethod(m, args);
- }
-
- /**
- * Creates a new {@link Batch.Call} instance that invokes a method
- * with the given parameters and returns the result.
- *
- * @param method the method reference to invoke
- * @param args zero or more arguments to be passed to the method
- * @param <T> the class type of the protocol implementation being invoked
- * @param <R> the return type for the method call
- * @return a {@code Callable} instance that will invoke the given method and
- * return the results
- * @see org.apache.hadoop.hbase.client.HTable#coprocessorExec(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback)
- */
- @Deprecated
- public static <T extends CoprocessorProtocol,R> Call<T,R> forMethod(
- final Method method, final Object... args) {
- return new Call<T,R>() {
- public R call(T instance) throws IOException {
- try {
- if (Proxy.isProxyClass(instance.getClass())) {
- InvocationHandler invoker = Proxy.getInvocationHandler(instance);
- return (R)invoker.invoke(instance, method, args);
- } else {
- LOG.warn("Non proxied invocation of method '"+method.getName()+"'!");
- return (R)method.invoke(instance, args);
- }
- }
- catch (IllegalAccessException iae) {
- throw new IOException("Unable to invoke method '"+
- method.getName()+"'", iae);
- }
- catch (InvocationTargetException ite) {
- throw new IOException(ite.toString(), ite);
- }
- catch (Throwable t) {
- throw new IOException(t.toString(), t);
- }
- }
- };
- }
-
/**
* Defines a unit of work to be executed.
*
* <p>
* When used with
- * {@link org.apache.hadoop.hbase.client.HTable#coprocessorExec(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback)}
+ * {@link org.apache.hadoop.hbase.client.HTable#coprocessorService(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call)}
* the implementations {@link Batch.Call#call(Object)} method will be invoked
* with a proxy to the
- * {@link org.apache.hadoop.hbase.ipc.CoprocessorProtocol}
+ * {@link org.apache.hadoop.hbase.coprocessor.CoprocessorService}
* sub-type instance.
* </p>
* @see org.apache.hadoop.hbase.client.coprocessor
- * @see org.apache.hadoop.hbase.client.HTable#coprocessorExec(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call)
- * @see org.apache.hadoop.hbase.client.HTable#coprocessorExec(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback)
+ * @see org.apache.hadoop.hbase.client.HTable#coprocessorService(byte[])
+ * @see org.apache.hadoop.hbase.client.HTable#coprocessorService(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call)
* @param <T> the instance type to be passed to
* {@link Batch.Call#call(Object)}
* @param <R> the return type from {@link Batch.Call#call(Object)}
@@ -157,15 +60,15 @@ public abstract class Batch {
*
* <p>
* When used with
- * {@link org.apache.hadoop.hbase.client.HTable#coprocessorExec(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback)},
+ * {@link org.apache.hadoop.hbase.client.HTable#coprocessorService(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call)}
* the implementation's {@link Batch.Callback#update(byte[], byte[], Object)}
* method will be called with the {@link Batch.Call#call(Object)} return value
* from each region in the selected range.
* </p>
* @param <R> the return type from the associated {@link Batch.Call#call(Object)}
- * @see org.apache.hadoop.hbase.client.HTable#coprocessorExec(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call, org.apache.hadoop.hbase.client.coprocessor.Batch.Callback)
+ * @see org.apache.hadoop.hbase.client.HTable#coprocessorService(Class, byte[], byte[], org.apache.hadoop.hbase.client.coprocessor.Batch.Call)
*/
public static interface Callback<R> {
public void update(byte[] region, byte[] row, R result);
}
-}
+}
\ No newline at end of file
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/LongColumnInterpreter.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/LongColumnInterpreter.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/LongColumnInterpreter.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/coprocessor/LongColumnInterpreter.java Wed Feb 13 20:58:23 2013
@@ -19,16 +19,15 @@
package org.apache.hadoop.hbase.client.coprocessor;
import java.io.IOException;
-import java.nio.ByteBuffer;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.coprocessor.ColumnInterpreter;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.EmptyMsg;
+import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.LongMsg;
import org.apache.hadoop.hbase.util.Bytes;
-import com.google.protobuf.ByteString;
-
/**
* a concrete column interpreter implementation. The cell value is a Long value
* and its promoted data type is also a Long value. For computing aggregation
@@ -39,7 +38,8 @@ import com.google.protobuf.ByteString;
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
-public class LongColumnInterpreter implements ColumnInterpreter<Long, Long> {
+public class LongColumnInterpreter extends ColumnInterpreter<Long, Long,
+ EmptyMsg, LongMsg, LongMsg> {
public Long getValue(byte[] colFamily, byte[] colQualifier, KeyValue kv)
throws IOException {
@@ -97,45 +97,40 @@ public class LongColumnInterpreter imple
return o;
}
-
@Override
- public Long parseResponseAsPromotedType(byte[] response) {
- ByteBuffer b = ByteBuffer.allocate(8).put(response);
- b.rewind();
- long l = b.getLong();
+ public Long castToCellType(Long l) {
return l;
}
@Override
- public Long castToCellType(Long l) {
- return l;
+ public EmptyMsg getRequestData() {
+ return EmptyMsg.getDefaultInstance();
}
@Override
- public ByteString columnInterpreterSpecificData() {
- // nothing
- return null;
+ public void initialize(EmptyMsg msg) {
+ //nothing
}
@Override
- public void initialize(ByteString bytes) {
- // nothing
+ public LongMsg getProtoForCellType(Long t) {
+ LongMsg.Builder builder = LongMsg.newBuilder();
+ return builder.setLongMsg(t).build();
}
@Override
- public ByteString getProtoForCellType(Long t) {
- return getProtoForPromotedOrCellType(t);
+ public LongMsg getProtoForPromotedType(Long s) {
+ LongMsg.Builder builder = LongMsg.newBuilder();
+ return builder.setLongMsg(s).build();
}
@Override
- public ByteString getProtoForPromotedType(Long s) {
- return getProtoForPromotedOrCellType(s);
+ public Long getPromotedValueFromProto(LongMsg r) {
+ return r.getLongMsg();
}
- private ByteString getProtoForPromotedOrCellType(Long s) {
- ByteBuffer bb = ByteBuffer.allocate(8).putLong(s);
- bb.rewind();
- ByteString bs = ByteString.copyFrom(bb);
- return bs;
+ @Override
+ public Long getCellValueFromProto(LongMsg q) {
+ return q.getLongMsg();
}
-}
\ No newline at end of file
+}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationAdmin.java Wed Feb 13 20:58:23 2013
@@ -185,7 +185,7 @@ public class ReplicationAdmin implements
boolean prev = true;
try {
prev = getReplicating();
- this.replicationZk.setReplicating(newState);
+ this.replicationZk.setReplication(newState);
} catch (KeeperException e) {
throw new IOException("Unable to set the replication state", e);
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/constraint/Constraints.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/constraint/Constraints.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/constraint/Constraints.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/constraint/Constraints.java Wed Feb 13 20:58:23 2013
@@ -133,7 +133,7 @@ public final class Constraints {
}
// now remove all the keys we found
for (ImmutableBytesWritable key : keys) {
- desc.remove(key.get());
+ desc.remove(key);
}
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/AggregateImplementation.java Wed Feb 13 20:58:23 2013
@@ -19,6 +19,10 @@
package org.apache.hadoop.hbase.coprocessor;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
@@ -42,6 +46,7 @@ import org.apache.hadoop.hbase.protobuf.
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
@@ -49,14 +54,20 @@ import com.google.protobuf.Service;
/**
* A concrete AggregateProtocol implementation. Its system level coprocessor
* that computes the aggregate function at a region level.
- * @param <T>
- * @param <S>
+ * {@link ColumnInterpreter} is used to interpret column value. This class is
+ * parameterized with the following (these are the types with which the {@link ColumnInterpreter}
+ * is parameterized, and for more description on these, refer to {@link ColumnInterpreter}):
+ * @param <T> Cell value data type
+ * @param <S> Promoted data type
+ * @param <P> PB message that is used to transport initializer specific bytes
+ * @param <Q> PB message that is used to transport Cell (<T>) instance
+ * @param <R> PB message that is used to transport Promoted (<S>) instance
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
-public class AggregateImplementation<T, S> extends AggregateService implements
- CoprocessorService, Coprocessor {
- protected static Log log = LogFactory.getLog(AggregateImplementation.class);
+public class AggregateImplementation<T, S, P extends Message, Q extends Message, R extends Message>
+extends AggregateService implements CoprocessorService, Coprocessor {
+ protected static final Log log = LogFactory.getLog(AggregateImplementation.class);
private RegionCoprocessorEnvironment env;
/**
@@ -73,7 +84,7 @@ public class AggregateImplementation<T,
AggregateResponse response = null;
T max = null;
try {
- ColumnInterpreter<T, S> ci = constructColumnInterpreterFromRequest(request);
+ ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
T temp;
Scan scan = ProtobufUtil.toScan(request.getScan());
scanner = env.getRegion().getScanner(scan);
@@ -96,7 +107,7 @@ public class AggregateImplementation<T,
} while (hasMoreRows);
if (max != null) {
AggregateResponse.Builder builder = AggregateResponse.newBuilder();
- builder.addFirstPart(ci.getProtoForCellType(max));
+ builder.addFirstPart(ci.getProtoForCellType(max).toByteString());
response = builder.build();
}
} catch (IOException e) {
@@ -127,7 +138,7 @@ public class AggregateImplementation<T,
InternalScanner scanner = null;
T min = null;
try {
- ColumnInterpreter<T, S> ci = constructColumnInterpreterFromRequest(request);
+ ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
T temp;
Scan scan = ProtobufUtil.toScan(request.getScan());
scanner = env.getRegion().getScanner(scan);
@@ -149,7 +160,7 @@ public class AggregateImplementation<T,
} while (hasMoreRows);
if (min != null) {
response = AggregateResponse.newBuilder().addFirstPart(
- ci.getProtoForCellType(min)).build();
+ ci.getProtoForCellType(min).toByteString()).build();
}
} catch (IOException e) {
ResponseConverter.setControllerException(controller, e);
@@ -179,7 +190,7 @@ public class AggregateImplementation<T,
InternalScanner scanner = null;
long sum = 0l;
try {
- ColumnInterpreter<T, S> ci = constructColumnInterpreterFromRequest(request);
+ ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null;
T temp;
Scan scan = ProtobufUtil.toScan(request.getScan());
@@ -203,7 +214,7 @@ public class AggregateImplementation<T,
} while (hasMoreRows);
if (sumVal != null) {
response = AggregateResponse.newBuilder().addFirstPart(
- ci.getProtoForPromotedType(sumVal)).build();
+ ci.getProtoForPromotedType(sumVal).toByteString()).build();
}
} catch (IOException e) {
ResponseConverter.setControllerException(controller, e);
@@ -287,7 +298,7 @@ public class AggregateImplementation<T,
AggregateResponse response = null;
InternalScanner scanner = null;
try {
- ColumnInterpreter<T, S> ci = constructColumnInterpreterFromRequest(request);
+ ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null;
Long rowCountVal = 0l;
Scan scan = ProtobufUtil.toScan(request.getScan());
@@ -311,7 +322,7 @@ public class AggregateImplementation<T,
rowCountVal++;
} while (hasMoreRows);
if (sumVal != null) {
- ByteString first = ci.getProtoForPromotedType(sumVal);
+ ByteString first = ci.getProtoForPromotedType(sumVal).toByteString();
AggregateResponse.Builder pair = AggregateResponse.newBuilder();
pair.addFirstPart(first);
ByteBuffer bb = ByteBuffer.allocate(8).putLong(rowCountVal);
@@ -346,7 +357,7 @@ public class AggregateImplementation<T,
InternalScanner scanner = null;
AggregateResponse response = null;
try {
- ColumnInterpreter<T, S> ci = constructColumnInterpreterFromRequest(request);
+ ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null, sumSqVal = null, tempVal = null;
long rowCountVal = 0l;
Scan scan = ProtobufUtil.toScan(request.getScan());
@@ -374,8 +385,8 @@ public class AggregateImplementation<T,
rowCountVal++;
} while (hasMoreRows);
if (sumVal != null) {
- ByteString first_sumVal = ci.getProtoForPromotedType(sumVal);
- ByteString first_sumSqVal = ci.getProtoForPromotedType(sumSqVal);
+ ByteString first_sumVal = ci.getProtoForPromotedType(sumVal).toByteString();
+ ByteString first_sumSqVal = ci.getProtoForPromotedType(sumSqVal).toByteString();
AggregateResponse.Builder pair = AggregateResponse.newBuilder();
pair.addFirstPart(first_sumVal);
pair.addFirstPart(first_sumSqVal);
@@ -410,7 +421,7 @@ public class AggregateImplementation<T,
AggregateResponse response = null;
InternalScanner scanner = null;
try {
- ColumnInterpreter<T, S> ci = constructColumnInterpreterFromRequest(request);
+ ColumnInterpreter<T, S, P, Q, R> ci = constructColumnInterpreterFromRequest(request);
S sumVal = null, sumWeights = null, tempVal = null, tempWeight = null;
Scan scan = ProtobufUtil.toScan(request.getScan());
scanner = env.getRegion().getScanner(scan);
@@ -442,9 +453,9 @@ public class AggregateImplementation<T,
sumVal = ci.add(sumVal, tempVal);
sumWeights = ci.add(sumWeights, tempWeight);
} while (hasMoreRows);
- ByteString first_sumVal = ci.getProtoForPromotedType(sumVal);
+ ByteString first_sumVal = ci.getProtoForPromotedType(sumVal).toByteString();
S s = sumWeights == null ? ci.castToReturnType(ci.getMinValue()) : sumWeights;
- ByteString first_sumWeights = ci.getProtoForPromotedType(s);
+ ByteString first_sumWeights = ci.getProtoForPromotedType(s).toByteString();
AggregateResponse.Builder pair = AggregateResponse.newBuilder();
pair.addFirstPart(first_sumVal);
pair.addFirstPart(first_sumWeights);
@@ -462,15 +473,17 @@ public class AggregateImplementation<T,
}
@SuppressWarnings("unchecked")
- ColumnInterpreter<T,S> constructColumnInterpreterFromRequest(
+ ColumnInterpreter<T,S,P,Q,R> constructColumnInterpreterFromRequest(
AggregateArgument request) throws IOException {
String className = request.getInterpreterClassName();
Class<?> cls;
try {
cls = Class.forName(className);
- ColumnInterpreter<T,S> ci = (ColumnInterpreter<T, S>) cls.newInstance();
+ ColumnInterpreter<T,S,P,Q,R> ci = (ColumnInterpreter<T, S, P, Q, R>) cls.newInstance();
if (request.hasInterpreterSpecificBytes()) {
- ci.initialize(request.getInterpreterSpecificBytes());
+ ByteString b = request.getInterpreterSpecificBytes();
+ P initMsg = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 2, b);
+ ci.initialize(initMsg);
}
return ci;
} catch (ClassNotFoundException e) {
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java Wed Feb 13 20:58:23 2013
@@ -140,7 +140,8 @@ public abstract class BaseRegionObserver
@Override
public InternalScanner preCompact(ObserverContext<RegionCoprocessorEnvironment> e,
- final HStore store, final InternalScanner scanner) throws IOException {
+ final HStore store, final InternalScanner scanner, final ScanType scanType)
+ throws IOException {
return scanner;
}
@@ -315,6 +316,12 @@ public abstract class BaseRegionObserver
}
@Override
+ public boolean postScannerFilterRow(final ObserverContext<RegionCoprocessorEnvironment> e,
+ final InternalScanner s, final byte[] currentRow, final boolean hasMore) throws IOException {
+ return hasMore;
+ }
+
+ @Override
public void preScannerClose(final ObserverContext<RegionCoprocessorEnvironment> e,
final InternalScanner s) throws IOException {
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/ColumnInterpreter.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/ColumnInterpreter.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/ColumnInterpreter.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/ColumnInterpreter.java Wed Feb 13 20:58:23 2013
@@ -24,9 +24,11 @@ import java.io.IOException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
/**
* Defines how value for specific column is interpreted and provides utility
@@ -35,7 +37,8 @@ import com.google.protobuf.ByteString;
* handle null case gracefully. Refer to {@link LongColumnInterpreter} for an
* example.
* <p>
- * Takes two generic parameters. The cell value type of the interpreter is <T>.
+ * Takes two generic parameters and three Message parameters.
+ * The cell value type of the interpreter is <T>.
* During some computations like sum, average, the return type can be different
* than the cell value data type, for eg, sum of int cell values might overflow
* in case of a int result, we should use Long for its result. Therefore, this
@@ -44,12 +47,19 @@ import com.google.protobuf.ByteString;
* <S>. There is a conversion method
* {@link ColumnInterpreter#castToReturnType(Object)} which takes a <T> type and
* returns a <S> type.
+ * The {@link AggregateImplementation} uses PB messages to initialize the
+ * user's ColumnInterpreter implementation, and for sending the responses
+ * back to {@link AggregationClient}.
* @param <T> Cell value data type
* @param <S> Promoted data type
+ * @param <P> PB message that is used to transport initializer specific bytes
+ * @param <Q> PB message that is used to transport Cell (<T>) instance
+ * @param <R> PB message that is used to transport Promoted (<S>) instance
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
-public interface ColumnInterpreter<T, S> {
+public abstract class ColumnInterpreter<T, S, P extends Message,
+Q extends Message, R extends Message> {
/**
* @param colFamily
@@ -58,7 +68,7 @@ public interface ColumnInterpreter<T, S>
* @return value of type T
* @throws IOException
*/
- T getValue(byte[] colFamily, byte[] colQualifier, KeyValue kv)
+ public abstract T getValue(byte[] colFamily, byte[] colQualifier, KeyValue kv)
throws IOException;
/**
@@ -67,36 +77,36 @@ public interface ColumnInterpreter<T, S>
* @return sum or non null value among (if either of them is null); otherwise
* returns a null.
*/
- public S add(S l1, S l2);
+ public abstract S add(S l1, S l2);
/**
* returns the maximum value for this type T
* @return max
*/
- T getMaxValue();
+ public abstract T getMaxValue();
- T getMinValue();
+ public abstract T getMinValue();
/**
* @param o1
* @param o2
* @return multiplication
*/
- S multiply(S o1, S o2);
+ public abstract S multiply(S o1, S o2);
/**
* @param o
* @return increment
*/
- S increment(S o);
+ public abstract S increment(S o);
/**
* provides casting opportunity between the data types.
* @param o
* @return cast
*/
- S castToReturnType(T o);
+ public abstract S castToReturnType(T o);
/**
* This takes care if either of arguments are null. returns 0 if they are
@@ -105,7 +115,7 @@ public interface ColumnInterpreter<T, S>
* <li>>0 if l1 > l2 or l1 is not null and l2 is null.
* <li>< 0 if l1 < l2 or l1 is null and l2 is not null.
*/
- int compare(final T l1, final T l2);
+ public abstract int compare(final T l1, final T l2);
/**
* used for computing average of <S> data values. Not providing the divide
@@ -114,51 +124,58 @@ public interface ColumnInterpreter<T, S>
* @param l
* @return Average
*/
- double divideForAvg(S o, Long l);
+ public abstract double divideForAvg(S o, Long l);
/**
* This method should return any additional data that is needed on the
* server side to construct the ColumnInterpreter. The server
- * will pass this to the {@link #initialize(ByteString)}
+ * will pass this to the {@link #initialize}
* method. If there is no ColumnInterpreter specific data (for e.g.,
* {@link LongColumnInterpreter}) then null should be returned.
* @return the PB message
*/
- ByteString columnInterpreterSpecificData();
+ public abstract P getRequestData();
/**
- * Return the PB for type T
+ * This method should initialize any field(s) of the ColumnInterpreter with
+ * a parsing of the passed message bytes (used on the server side).
+ * @param msg
+ */
+ public abstract void initialize(P msg);
+
+ /**
+ * This method gets the PB message corresponding to the cell type
* @param t
- * @return PB-message
+ * @return the PB message for the cell-type instance
*/
- ByteString getProtoForCellType(T t);
+ public abstract Q getProtoForCellType(T t);
/**
- * Return the PB for type S
- * @param s
- * @return PB-message
+ * This method gets the PB message corresponding to the cell type
+ * @param q
+ * @return the cell-type instance from the PB message
*/
- ByteString getProtoForPromotedType(S s);
+ public abstract T getCellValueFromProto(Q q);
/**
- * This method should initialize any field(s) of the ColumnInterpreter with
- * a parsing of the passed message bytes (used on the server side).
- * @param bytes
+ * This method gets the PB message corresponding to the promoted type
+ * @param s
+ * @return the PB message for the promoted-type instance
*/
- void initialize(ByteString bytes);
-
+ public abstract R getProtoForPromotedType(S s);
+
/**
- * Converts the bytes in the server's response to the expected type S
- * @param response
- * @return response of type S constructed from the message
+ * This method gets the promoted type from the proto message
+ * @param r
+ * @return the promoted-type instance from the PB message
*/
- S parseResponseAsPromotedType(byte[] response);
-
+ public abstract S getPromotedValueFromProto(R r);
+
/**
* The response message comes as type S. This will convert/cast it to T.
* In some sense, performs the opposite of {@link #castToReturnType(Object)}
* @param response
* @return cast
*/
- T castToCellType(S response);
+ public abstract T castToCellType(S response);
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/CoprocessorHost.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/CoprocessorHost.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/CoprocessorHost.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/CoprocessorHost.java Wed Feb 13 20:58:23 2013
@@ -36,7 +36,6 @@ import org.apache.hadoop.hbase.HBaseConf
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
-import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.SortedCopyOnWriteSet;
@@ -48,6 +47,8 @@ import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URL;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.*;
import java.util.concurrent.ConcurrentMap;
import java.util.jar.JarEntry;
@@ -73,6 +74,10 @@ public abstract class CoprocessorHost<E
public static final String WAL_COPROCESSOR_CONF_KEY =
"hbase.coprocessor.wal.classes";
+ //coprocessor jars are put under ${hbase.local.dir}/coprocessor/jars/
+ private static final String COPROCESSOR_JARS_DIR = File.separator
+ + "coprocessor" + File.separator + "jars" + File.separator;
+
private static final Log LOG = LogFactory.getLog(CoprocessorHost.class);
/** Ordered set of loaded coprocessors with lock */
protected SortedSet<E> coprocessors =
@@ -131,7 +136,7 @@ public abstract class CoprocessorHost<E
protected void loadSystemCoprocessors(Configuration conf, String confKey) {
Class<?> implClass = null;
- // load default coprocessors from configure file
+ // load default coprocessors from configure file
String[] defaultCPClasses = conf.getStrings(confKey);
if (defaultCPClasses == null || defaultCPClasses.length == 0)
return;
@@ -175,7 +180,7 @@ public abstract class CoprocessorHost<E
public E load(Path path, String className, int priority,
Configuration conf) throws IOException {
Class<?> implClass = null;
- LOG.debug("Loading coprocessor class " + className + " with path " +
+ LOG.debug("Loading coprocessor class " + className + " with path " +
path + " and priority " + priority);
ClassLoader cl = null;
@@ -210,13 +215,13 @@ public abstract class CoprocessorHost<E
if (!path.toString().endsWith(".jar")) {
throw new IOException(path.toString() + ": not a jar file?");
}
- FileSystem fs = path.getFileSystem(HBaseConfiguration.create());
- Path dst = new Path(System.getProperty("java.io.tmpdir") +
- java.io.File.separator +"." + pathPrefix +
+ FileSystem fs = path.getFileSystem(this.conf);
+ File parentDir = new File(this.conf.get("hbase.local.dir") + COPROCESSOR_JARS_DIR);
+ parentDir.mkdirs();
+ File dst = new File(parentDir, "." + pathPrefix +
"." + className + "." + System.currentTimeMillis() + ".jar");
- fs.copyToLocalFile(path, dst);
- File tmpLocal = new File(dst.toString());
- tmpLocal.deleteOnExit();
+ fs.copyToLocalFile(path, new Path(dst.toString()));
+ dst.deleteOnExit();
// TODO: code weaving goes here
@@ -229,8 +234,8 @@ public abstract class CoprocessorHost<E
// NOTE: Path.toURL is deprecated (toURI instead) but the URLClassLoader
// unsurprisingly wants URLs, not URIs; so we will use the deprecated
// method which returns URLs for as long as it is available
- List<URL> paths = new ArrayList<URL>();
- URL url = new File(dst.toString()).getCanonicalFile().toURL();
+ final List<URL> paths = new ArrayList<URL>();
+ URL url = dst.getCanonicalFile().toURL();
paths.add(url);
JarFile jarFile = new JarFile(dst.toString());
@@ -238,8 +243,7 @@ public abstract class CoprocessorHost<E
while (entries.hasMoreElements()) {
JarEntry entry = entries.nextElement();
if (entry.getName().matches("/lib/[^/]+\\.jar")) {
- File file = new File(System.getProperty("java.io.tmpdir") +
- java.io.File.separator +"." + pathPrefix +
+ File file = new File(parentDir, "." + pathPrefix +
"." + className + "." + System.currentTimeMillis() + "." + entry.getName().substring(5));
IOUtils.copyBytes(jarFile.getInputStream(entry), new FileOutputStream(file), conf, true);
file.deleteOnExit();
@@ -248,7 +252,13 @@ public abstract class CoprocessorHost<E
}
jarFile.close();
- cl = new CoprocessorClassLoader(paths, this.getClass().getClassLoader());
+ cl = AccessController.doPrivileged(new PrivilegedAction<CoprocessorClassLoader>() {
+ @Override
+ public CoprocessorClassLoader run() {
+ return new CoprocessorClassLoader(paths, this.getClass().getClassLoader());
+ }
+ });
+
// cache cp classloader as a weak value, will be GC'ed when no reference left
ClassLoader prev = classLoadersCache.putIfAbsent(path, cl);
if (prev != null) {
@@ -470,6 +480,10 @@ public abstract class CoprocessorHost<E
return table.exists(get);
}
+ public Boolean[] exists(List<Get> gets) throws IOException{
+ return table.exists(gets);
+ }
+
public void put(Put put) throws IOException {
table.put(put);
}
@@ -547,16 +561,6 @@ public abstract class CoprocessorHost<E
return tableName;
}
- public RowLock lockRow(byte[] row) throws IOException {
- throw new RuntimeException(
- "row locking is not allowed within the coprocessor environment");
- }
-
- public void unlockRow(RowLock rl) throws IOException {
- throw new RuntimeException(
- "row locking is not allowed within the coprocessor environment");
- }
-
@Override
public void batch(List<? extends Row> actions, Object[] results)
throws IOException, InterruptedException {
@@ -587,26 +591,6 @@ public abstract class CoprocessorHost<E
}
@Override
- public <T extends CoprocessorProtocol, R> void coprocessorExec(Class<T> protocol,
- byte[] startKey, byte[] endKey, Batch.Call<T, R> callable,
- Batch.Callback<R> callback) throws IOException, Throwable {
- table.coprocessorExec(protocol, startKey, endKey, callable, callback);
- }
-
- @Override
- public <T extends CoprocessorProtocol, R> Map<byte[], R> coprocessorExec(
- Class<T> protocol, byte[] startKey, byte[] endKey, Batch.Call<T, R> callable)
- throws IOException, Throwable {
- return table.coprocessorExec(protocol, startKey, endKey, callable);
- }
-
- @Override
- public <T extends CoprocessorProtocol> T coprocessorProxy(Class<T> protocol,
- byte[] row) {
- return table.coprocessorProxy(protocol, row);
- }
-
- @Override
public CoprocessorRpcChannel coprocessorService(byte[] row) {
return table.coprocessorService(row);
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java Wed Feb 13 20:58:23 2013
@@ -167,12 +167,14 @@ public interface RegionObserver extends
* @param store the store being compacted
* @param scanner the scanner over existing data used in the store file
* rewriting
+ * @param scanType type of Scan
* @return the scanner to use during compaction. Should not be {@code null}
* unless the implementation is writing new store files on its own.
* @throws IOException if an error occurred on the coprocessor
*/
InternalScanner preCompact(final ObserverContext<RegionCoprocessorEnvironment> c,
- final HStore store, final InternalScanner scanner) throws IOException;
+ final HStore store, final InternalScanner scanner,
+ final ScanType scanType) throws IOException;
/**
* Called prior to writing the {@link StoreFile}s selected for compaction into
@@ -735,6 +737,27 @@ public interface RegionObserver extends
throws IOException;
/**
+ * This will be called by the scan flow when the current scanned row is being filtered out by the
+ * filter. The filter may be filtering out the row via any of the below scenarios
+ * <ol>
+ * <li>
+ * <code>boolean filterRowKey(byte [] buffer, int offset, int length)</code> returning true</li>
+ * <li>
+ * <code>boolean filterRow()</code> returning true</li>
+ * <li>
+ * <code>void filterRow(List<KeyValue> kvs)</code> removing all the kvs from the passed List</li>
+ * </ol>
+ * @param c the environment provided by the region server
+ * @param s the scanner
+ * @param currentRow The current rowkey which got filtered out
+ * @param hasMore the 'has more' indication
+ * @return whether more rows are available for the scanner or not
+ * @throws IOException
+ */
+ boolean postScannerFilterRow(final ObserverContext<RegionCoprocessorEnvironment> c,
+ final InternalScanner s, final byte[] currentRow, final boolean hasMore) throws IOException;
+
+ /**
* Called before the client closes a scanner.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/executor/EventHandler.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/executor/EventHandler.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/executor/EventHandler.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/executor/EventHandler.java Wed Feb 13 20:58:23 2013
@@ -67,7 +67,7 @@ public abstract class EventHandler imple
protected Server server;
// sequence id generator for default FIFO ordering of events
- protected static AtomicLong seqids = new AtomicLong(0);
+ protected static final AtomicLong seqids = new AtomicLong(0);
// sequence id for this event
private final long seqid;
@@ -120,7 +120,7 @@ public abstract class EventHandler imple
// Messages originating from Master to RS
M_RS_OPEN_REGION (20, ExecutorType.RS_OPEN_REGION), // Master asking RS to open a region
- M_RS_OPEN_ROOT (21, ExecutorType.RS_OPEN_REGION), // Master asking RS to open root
+ M_RS_OPEN_ROOT (21, ExecutorType.RS_OPEN_ROOT), // Master asking RS to open root
M_RS_OPEN_META (22, ExecutorType.RS_OPEN_META), // Master asking RS to open meta
M_RS_CLOSE_REGION (23, ExecutorType.RS_CLOSE_REGION), // Master asking RS to close a region
M_RS_CLOSE_ROOT (24, ExecutorType.RS_CLOSE_ROOT), // Master asking RS to close root
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/ByteArrayComparable.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/ByteArrayComparable.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/ByteArrayComparable.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/ByteArrayComparable.java Wed Feb 13 20:58:23 2013
@@ -75,11 +75,10 @@ public abstract class ByteArrayComparabl
* @return true if and only if the fields of the comparator that are serialized
* are equal to the corresponding fields in other. Used for testing.
*/
- boolean areSerializedFieldsEqual(ByteArrayComparable o) {
- if (o == this) return true;
- if (!(o instanceof ByteArrayComparable)) return false;
+ boolean areSerializedFieldsEqual(ByteArrayComparable other) {
+ if (other == this) return true;
- return Bytes.equals(this.getValue(), o.getValue());
+ return Bytes.equals(this.getValue(), other.getValue());
}
@Override
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/CompareFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/CompareFilter.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/CompareFilter.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/CompareFilter.java Wed Feb 13 20:58:23 2013
@@ -156,7 +156,7 @@ public abstract class CompareFilter exte
/**
*
- * @param other
+ * @param o
* @return true if and only if the fields of the filter that are serialized
* are equal to the corresponding fields in other. Used for testing.
*/
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/DependentColumnFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/DependentColumnFilter.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/DependentColumnFilter.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/DependentColumnFilter.java Wed Feb 13 20:58:23 2013
@@ -261,10 +261,12 @@ public class DependentColumnFilter exten
}
/**
- * @param other
+ * @param o
* @return true if and only if the fields of the filter that are serialized
* are equal to the corresponding fields in other. Used for testing.
*/
+ @edu.umd.cs.findbugs.annotations.SuppressWarnings(
+ value="RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE")
boolean areSerializedFieldsEqual(Filter o) {
if (o == this) return true;
if (!(o instanceof DependentColumnFilter)) return false;
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/Filter.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/Filter.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/Filter.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/Filter.java Wed Feb 13 20:58:23 2013
@@ -172,6 +172,14 @@ public abstract class Filter {
abstract public KeyValue getNextKeyHint(final KeyValue currentKV);
/**
+ * Check that given column family is essential for filter to check row. Most
+ * filters always return true here. But some could have more sophisticated
+ * logic which could significantly reduce scanning process by not even
+ * touching columns until we are 100% sure that it's data is needed in result.
+ */
+ abstract public boolean isFamilyEssential(byte[] name);
+
+ /**
* @return The filter serialized using pb
*/
abstract public byte [] toByteArray();
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterBase.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterBase.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterBase.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterBase.java Wed Feb 13 20:58:23 2013
@@ -134,6 +134,16 @@ public abstract class FilterBase extends
}
/**
+ * By default, we require all scan's column families to be present. Our
+ * subclasses may be more precise.
+ *
+ * @inheritDoc
+ */
+ public boolean isFamilyEssential(byte[] name) {
+ return true;
+ }
+
+ /**
* Given the filter's arguments it constructs the filter
* <p>
* @param filterArguments the filter's arguments
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterList.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterList.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterList.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterList.java Wed Feb 13 20:58:23 2013
@@ -361,6 +361,16 @@ public class FilterList extends Filter {
}
@Override
+ public boolean isFamilyEssential(byte[] name) {
+ for (Filter filter : filters) {
+ if (filter.isFamilyEssential(name)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
public String toString() {
return toString(MAX_LOG_FILTERS);
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterWrapper.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterWrapper.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterWrapper.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FilterWrapper.java Wed Feb 13 20:58:23 2013
@@ -136,6 +136,11 @@ public class FilterWrapper extends Filte
}
}
+ @Override
+ public boolean isFamilyEssential(byte[] name) {
+ return filter.isFamilyEssential(name);
+ };
+
/**
* @param other
* @return true if and only if the fields of the filter that are serialized
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyOnlyFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyOnlyFilter.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyOnlyFilter.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyOnlyFilter.java Wed Feb 13 20:58:23 2013
@@ -90,13 +90,13 @@ public class FirstKeyOnlyFilter extends
*/
public static FirstKeyOnlyFilter parseFrom(final byte [] pbBytes)
throws DeserializationException {
- FilterProtos.FirstKeyOnlyFilter proto;
+ // There is nothing to deserialize. Why do this at all?
try {
- proto = FilterProtos.FirstKeyOnlyFilter.parseFrom(pbBytes);
+ FilterProtos.FirstKeyOnlyFilter.parseFrom(pbBytes);
} catch (InvalidProtocolBufferException e) {
throw new DeserializationException(e);
}
-
+ // Just return a new instance.
return new FirstKeyOnlyFilter();
}
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyValueMatchingQualifiersFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyValueMatchingQualifiersFilter.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyValueMatchingQualifiersFilter.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/FirstKeyValueMatchingQualifiersFilter.java Wed Feb 13 20:58:23 2013
@@ -18,20 +18,17 @@
package org.apache.hadoop.hbase.filter;
-import java.util.Collections;
-import java.util.Set;
-import java.util.TreeSet;
-
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hbase.DeserializationException;
import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.mapreduce.RowCounter;
import org.apache.hadoop.hbase.protobuf.generated.FilterProtos;
import org.apache.hadoop.hbase.util.Bytes;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.InvalidProtocolBufferException;
+import java.util.Set;
+import java.util.TreeSet;
/**
* The filter looks for the given columns in KeyValue. Once there is a match for
@@ -40,7 +37,8 @@ import com.google.protobuf.InvalidProtoc
* <p>
* Note : It may emit KVs which do not have the given columns in them, if
* these KVs happen to occur before a KV which does have a match. Given this
- * caveat, this filter is only useful for special cases like {@link RowCounter}.
+ * caveat, this filter is only useful for special cases
+ * like {@link org.apache.hadoop.hbase.mapreduce.RowCounter}.
* <p>
*/
@InterfaceAudience.Public
Modified: hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/NullComparator.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/NullComparator.java?rev=1445918&r1=1445917&r2=1445918&view=diff
==============================================================================
--- hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/NullComparator.java (original)
+++ hbase/branches/hbase-7290/hbase-server/src/main/java/org/apache/hadoop/hbase/filter/NullComparator.java Wed Feb 13 20:58:23 2013
@@ -44,6 +44,11 @@ public class NullComparator extends Byte
}
@Override
+ public boolean equals(Object obj) {
+ return obj == null;
+ }
+
+ @Override
public int compareTo(byte[] value, int offset, int length) {
throw new UnsupportedOperationException();
}