You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by je...@apache.org on 2014/02/11 01:32:26 UTC
[08/13] Port Phoenix to Hbase0.98
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/53f7d3ce/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
index 325c3a7..54eb014 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataProtocol.java
@@ -19,25 +19,22 @@
*/
package org.apache.phoenix.coprocessor;
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
import java.util.Collections;
import java.util.List;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableUtils;
-
-import com.google.common.collect.Lists;
+import org.apache.phoenix.coprocessor.generated.MetaDataProtos;
+import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataResponse;
+import org.apache.phoenix.coprocessor.generated.MetaDataProtos.MetaDataService;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableImpl;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.MetaDataUtil;
+import com.google.common.collect.Lists;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.HBaseZeroCopyByteString;
+
/**
*
@@ -56,7 +53,7 @@ import org.apache.phoenix.util.MetaDataUtil;
*
* @since 0.1
*/
-public interface MetaDataProtocol extends CoprocessorProtocol {
+public abstract class MetaDataProtocol extends MetaDataService {
public static final int PHOENIX_MAJOR_VERSION = 3;
public static final int PHOENIX_MINOR_VERSION = 0;
public static final int PHOENIX_PATCH_NUMBER = 0;
@@ -92,7 +89,7 @@ public interface MetaDataProtocol extends CoprocessorProtocol {
PARENT_TABLE_NOT_FOUND
};
- public static class MetaDataMutationResult implements Writable {
+ public static class MetaDataMutationResult {
private MutationCode returnCode;
private long mutationTime;
private PTable table;
@@ -156,114 +153,54 @@ public interface MetaDataProtocol extends CoprocessorProtocol {
return familyName;
}
- @Override
- public void readFields(DataInput input) throws IOException {
- this.returnCode = MutationCode.values()[WritableUtils.readVInt(input)];
- this.mutationTime = input.readLong();
- wasUpdated = input.readBoolean();
- if (wasUpdated) {
- this.table = new PTableImpl();
- this.table.readFields(input);
- }
- columnName = Bytes.readByteArray(input);
- if (columnName.length > 0) {
- familyName = Bytes.readByteArray(input);
- }
- boolean hasTablesToDelete = input.readBoolean();
- if (hasTablesToDelete) {
- int count = input.readInt();
- tableNamesToDelete = Lists.newArrayListWithExpectedSize(count);
- for( int i = 0 ; i < count ; i++ ){
- byte[] tableName = Bytes.readByteArray(input);
- tableNamesToDelete.add(tableName);
- }
+ public static MetaDataMutationResult constructFromProto(MetaDataResponse proto) {
+ MetaDataMutationResult result = new MetaDataMutationResult();
+ result.returnCode = MutationCode.values()[proto.getReturnCode().ordinal()];
+ result.mutationTime = proto.getMutationTime();
+ if (proto.hasTable()) {
+ result.wasUpdated = true;
+ result.table = PTableImpl.createFromProto(proto.getTable());
+ }
+ if (proto.getTablesToDeleteCount() > 0) {
+ result.tableNamesToDelete =
+ Lists.newArrayListWithExpectedSize(proto.getTablesToDeleteCount());
+ for (ByteString tableName : proto.getTablesToDeleteList()) {
+ result.tableNamesToDelete.add(tableName.toByteArray());
}
+ }
+ result.columnName = ByteUtil.EMPTY_BYTE_ARRAY;
+ if(proto.hasColumnName()){
+ result.columnName = proto.getColumnName().toByteArray();
+ }
+ if(proto.hasFamilyName()){
+ result.familyName = proto.getFamilyName().toByteArray();
+ }
+ return result;
}
-
- @Override
- public void write(DataOutput output) throws IOException {
- WritableUtils.writeVInt(output, returnCode.ordinal());
- output.writeLong(mutationTime);
- output.writeBoolean(table != null);
- if (table != null) {
- table.write(output);
+
+ public static MetaDataResponse toProto(MetaDataMutationResult result) {
+ MetaDataProtos.MetaDataResponse.Builder builder =
+ MetaDataProtos.MetaDataResponse.newBuilder();
+ if (result != null) {
+ builder.setReturnCode(MetaDataProtos.MutationCode.values()[result.getMutationCode()
+ .ordinal()]);
+ builder.setMutationTime(result.getMutationTime());
+ if (result.table != null) {
+ builder.setTable(PTableImpl.toProto(result.table));
}
- Bytes.writeByteArray(output, columnName == null ? ByteUtil.EMPTY_BYTE_ARRAY : columnName);
- if (columnName != null) {
- Bytes.writeByteArray(output, familyName == null ? ByteUtil.EMPTY_BYTE_ARRAY : familyName);
+ if (result.getTableNamesToDelete() != null) {
+ for (byte[] tableName : result.tableNamesToDelete) {
+ builder.addTablesToDelete(HBaseZeroCopyByteString.wrap(tableName));
+ }
}
- if(tableNamesToDelete != null && tableNamesToDelete.size() > 0 ) {
- output.writeBoolean(true);
- output.writeInt(tableNamesToDelete.size());
- for(byte[] tableName : tableNamesToDelete) {
- Bytes.writeByteArray(output,tableName);
- }
-
- } else {
- output.writeBoolean(false);
+ if(result.getColumnName() != null){
+ builder.setColumnName(HBaseZeroCopyByteString.wrap(result.getColumnName()));
}
-
+ if(result.getFamilyName() != null){
+ builder.setFamilyName(HBaseZeroCopyByteString.wrap(result.getFamilyName()));
+ }
+ }
+ return builder.build();
}
}
-
- /**
- * The the latest Phoenix table at or before the given clientTimestamp. If the
- * client already has the latest (based on the tableTimestamp), then no table
- * is returned.
- * @param tenantId
- * @param schemaName
- * @param tableName
- * @param tableTimestamp
- * @param clientTimestamp
- * @return MetaDataMutationResult
- * @throws IOException
- */
- MetaDataMutationResult getTable(byte[] tenantId, byte[] schemaName, byte[] tableName, long tableTimestamp, long clientTimestamp) throws IOException;
-
- /**
- * Create a new Phoenix table
- * @param tableMetadata
- * @return MetaDataMutationResult
- * @throws IOException
- */
- MetaDataMutationResult createTable(List<Mutation> tableMetadata) throws IOException;
-
- /**
- * Drop an existing Phoenix table
- * @param tableMetadata
- * @param tableType
- * @return MetaDataMutationResult
- * @throws IOException
- */
- MetaDataMutationResult dropTable(List<Mutation> tableMetadata, String tableType) throws IOException;
-
- /**
- * Add a column to an existing Phoenix table
- * @param tableMetadata
- * @return MetaDataMutationResult
- * @throws IOException
- */
- MetaDataMutationResult addColumn(List<Mutation> tableMetadata) throws IOException;
-
- /**
- * Drop a column from an existing Phoenix table
- * @param tableMetadata
- * @return MetaDataMutationResult
- * @throws IOException
- */
- MetaDataMutationResult dropColumn(List<Mutation> tableMetadata) throws IOException;
-
- MetaDataMutationResult updateIndexState(List<Mutation> tableMetadata) throws IOException;
-
- /**
- * Clears the server-side cache of table meta data. Used between test runs to
- * ensure no side effects.
- */
- void clearCache();
-
- /**
- * Get the version of the server-side HBase and phoenix.jar. Used when initially connecting
- * to a cluster to ensure that the client and server jars are compatible.
- */
- long getVersion();
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/53f7d3ce/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
index 543cde0..a153e73 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
@@ -26,6 +26,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.util.List;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Scan;
@@ -186,7 +187,7 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
@Override
- public boolean next(List<KeyValue> results) throws IOException {
+ public boolean next(List<Cell> results) throws IOException {
try {
if (isFilterDone()) {
return false;
@@ -211,6 +212,11 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
} finally {
chunk.close(); }
}
+
+ @Override
+ public long getMaxResultSize() {
+ return s.getMaxResultSize();
+ }
};
}
@@ -224,7 +230,7 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
return new RegionScanner() {
@Override
- public boolean next(List<KeyValue> results) throws IOException {
+ public boolean next(List<Cell> results) throws IOException {
try {
return s.next(results);
} catch (Throwable t) {
@@ -234,17 +240,7 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
@Override
- public boolean next(List<KeyValue> results, String metric) throws IOException {
- try {
- return s.next(results, metric);
- } catch (Throwable t) {
- ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionNameAsString(), t);
- return false; // impossible
- }
- }
-
- @Override
- public boolean next(List<KeyValue> result, int limit) throws IOException {
+ public boolean next(List<Cell> result, int limit) throws IOException {
try {
return s.next(result, limit);
} catch (Throwable t) {
@@ -254,16 +250,6 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
@Override
- public boolean next(List<KeyValue> result, int limit, String metric) throws IOException {
- try {
- return s.next(result, limit, metric);
- } catch (Throwable t) {
- ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionNameAsString(), t);
- return false; // impossible
- }
- }
-
- @Override
public void close() throws IOException {
s.close();
}
@@ -274,7 +260,7 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
@Override
- public boolean isFilterDone() {
+ public boolean isFilterDone() throws IOException {
return s.isFilterDone();
}
@@ -289,9 +275,9 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
@Override
- public boolean nextRaw(List<KeyValue> result, String metric) throws IOException {
+ public boolean nextRaw(List<Cell> result) throws IOException {
try {
- return s.nextRaw(result, metric);
+ return s.nextRaw(result);
} catch (Throwable t) {
ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionNameAsString(), t);
return false; // impossible
@@ -299,14 +285,19 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
@Override
- public boolean nextRaw(List<KeyValue> result, int limit, String metric) throws IOException {
+ public boolean nextRaw(List<Cell> result, int limit) throws IOException {
try {
- return s.nextRaw(result, limit, metric);
+ return s.nextRaw(result, limit);
} catch (Throwable t) {
ServerUtil.throwIOException(c.getEnvironment().getRegion().getRegionNameAsString(), t);
return false; // impossible
}
}
+
+ @Override
+ public long getMaxResultSize() {
+ return s.getMaxResultSize();
+ }
};
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/53f7d3ce/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
index f1bebd4..e82d4ee 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
@@ -4,8 +4,9 @@ import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.NavigableMap;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Append;
@@ -20,10 +21,9 @@ import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.TimeRange;
import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegion.RowLock;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.apache.hadoop.hbase.util.Pair;
-
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.QueryConstants;
@@ -34,6 +34,8 @@ import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.ServerUtil;
+import com.google.common.collect.Lists;
+
/**
*
* Region observer coprocessor for sequence operations:
@@ -58,11 +60,21 @@ public class SequenceRegionObserver extends BaseRegionObserver {
private static Result getErrorResult(byte[] row, long timestamp, int errorCode) {
byte[] errorCodeBuf = new byte[PDataType.INTEGER.getByteSize()];
PDataType.INTEGER.getCodec().encodeInt(errorCode, errorCodeBuf, 0);
- return new Result(Collections.singletonList(
- KeyValueUtil.newKeyValue(row,
+ return Result.create(Collections.singletonList(
+ (Cell)KeyValueUtil.newKeyValue(row,
PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES,
QueryConstants.EMPTY_COLUMN_BYTES, timestamp, errorCodeBuf)));
}
+
+ private static void acquireLock(HRegion region, byte[] key, List<RowLock> locks)
+ throws IOException {
+ RowLock rowLock = region.getRowLock(key);
+ if (rowLock == null) {
+ throw new IOException("Failed to acquire lock on " + Bytes.toStringBinary(key));
+ }
+ locks.add(rowLock);
+ }
+
/**
*
* Use PreIncrement hook of BaseRegionObserver to overcome deficiencies in Increment
@@ -83,10 +95,11 @@ public class SequenceRegionObserver extends BaseRegionObserver {
e.complete();
HRegion region = env.getRegion();
byte[] row = increment.getRow();
+ List<RowLock> locks = Lists.newArrayList();
TimeRange tr = increment.getTimeRange();
region.startRegionOperation();
try {
- Integer lid = region.getLock(null, row, true);
+ acquireLock(region, row, locks);
try {
long maxTimestamp = tr.getMax();
if (maxTimestamp == HConstants.LATEST_TIMESTAMP) {
@@ -95,10 +108,10 @@ public class SequenceRegionObserver extends BaseRegionObserver {
}
Get get = new Get(row);
get.setTimeRange(tr.getMin(), tr.getMax());
- for (Map.Entry<byte[],NavigableMap<byte[], Long>> entry : increment.getFamilyMap().entrySet()) {
+ for (Map.Entry<byte[], List<Cell>> entry : increment.getFamilyCellMap().entrySet()) {
byte[] cf = entry.getKey();
- for (byte[] cq : entry.getValue().keySet()) {
- get.addColumn(cf, cq);
+ for (Cell cq : entry.getValue()) {
+ get.addColumn(cf, CellUtil.cloneQualifier(cq));
}
}
Result result = region.get(get);
@@ -108,24 +121,26 @@ public class SequenceRegionObserver extends BaseRegionObserver {
KeyValue currentValueKV = Sequence.getCurrentValueKV(result);
KeyValue incrementByKV = Sequence.getIncrementByKV(result);
KeyValue cacheSizeKV = Sequence.getCacheSizeKV(result);
- long value = PDataType.LONG.getCodec().decodeLong(currentValueKV.getBuffer(), currentValueKV.getValueOffset(), null);
- long incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getBuffer(), incrementByKV.getValueOffset(), null);
- int cacheSize = PDataType.INTEGER.getCodec().decodeInt(cacheSizeKV.getBuffer(), cacheSizeKV.getValueOffset(), null);
+ long value = PDataType.LONG.getCodec().decodeLong(currentValueKV.getValueArray(), currentValueKV.getValueOffset(), null);
+ long incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getValueArray(), incrementByKV.getValueOffset(), null);
+ int cacheSize = PDataType.INTEGER.getCodec().decodeInt(cacheSizeKV.getValueArray(), cacheSizeKV.getValueOffset(), null);
value += incrementBy * cacheSize;
byte[] valueBuffer = new byte[PDataType.LONG.getByteSize()];
PDataType.LONG.getCodec().encodeLong(value, valueBuffer, 0);
Put put = new Put(row, currentValueKV.getTimestamp());
// Hold timestamp constant for sequences, so that clients always only see the latest value
// regardless of when they connect.
- KeyValue newCurrentValueKV = KeyValueUtil.newKeyValue(row, currentValueKV.getFamily(), currentValueKV.getQualifier(), currentValueKV.getTimestamp(), valueBuffer);
+ KeyValue newCurrentValueKV = KeyValueUtil.newKeyValue(row, 0, row.length,
+ currentValueKV.getFamilyArray(), currentValueKV.getFamilyOffset(), currentValueKV.getFamilyLength(),
+ currentValueKV.getQualifierArray(), currentValueKV.getQualifierOffset(), currentValueKV.getQualifierLength(),
+ currentValueKV.getTimestamp(), valueBuffer, 0, valueBuffer.length);
+
put.add(newCurrentValueKV);
- @SuppressWarnings("unchecked")
- Pair<Mutation,Integer>[] mutations = new Pair[1];
- mutations[0] = new Pair<Mutation,Integer>(put, lid);
+ Mutation[] mutations = new Mutation[]{put};
region.batchMutate(mutations);
return Sequence.replaceCurrentValueKV(result, newCurrentValueKV);
} finally {
- region.releaseRowLock(lid);
+ region.releaseRowLocks(locks);
}
} catch (Throwable t) {
ServerUtil.throwIOException("Increment of sequence " + Bytes.toStringBinary(row), t);
@@ -149,7 +164,7 @@ public class SequenceRegionObserver extends BaseRegionObserver {
return null;
}
Op op = Op.values()[opBuf[0]];
- KeyValue keyValue = append.getFamilyMap().values().iterator().next().iterator().next();
+ Cell keyValue = append.getFamilyCellMap().values().iterator().next().iterator().next();
long clientTimestamp = HConstants.LATEST_TIMESTAMP;
long minGetTimestamp = MetaDataProtocol.MIN_TABLE_TIMESTAMP;
@@ -187,12 +202,13 @@ public class SequenceRegionObserver extends BaseRegionObserver {
e.complete();
HRegion region = env.getRegion();
byte[] row = append.getRow();
+ List<RowLock> locks = Lists.newArrayList();
region.startRegionOperation();
try {
- Integer lid = region.getLock(null, row, true);
+ acquireLock(region, row, locks);
try {
- byte[] family = keyValue.getFamily();
- byte[] qualifier = keyValue.getQualifier();
+ byte[] family = CellUtil.cloneFamily(keyValue);
+ byte[] qualifier = CellUtil.cloneQualifier(keyValue);
Get get = new Get(row);
get.setTimeRange(minGetTimestamp, maxGetTimestamp);
@@ -212,40 +228,42 @@ public class SequenceRegionObserver extends BaseRegionObserver {
case RETURN_SEQUENCE:
KeyValue currentValueKV = result.raw()[0];
long expectedValue = PDataType.LONG.getCodec().decodeLong(append.getAttribute(CURRENT_VALUE_ATTRIB), 0, null);
- long value = PDataType.LONG.getCodec().decodeLong(currentValueKV.getBuffer(), currentValueKV.getValueOffset(), null);
+ long value = PDataType.LONG.getCodec().decodeLong(currentValueKV.getValueArray(),
+ currentValueKV.getValueOffset(), null);
// Timestamp should match exactly, or we may have the wrong sequence
if (expectedValue != value || currentValueKV.getTimestamp() != clientTimestamp) {
- return new Result(Collections.singletonList(KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, currentValueKV.getTimestamp(), ByteUtil.EMPTY_BYTE_ARRAY)));
+ return Result.create(Collections.singletonList(
+ (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES,
+ QueryConstants.EMPTY_COLUMN_BYTES, currentValueKV.getTimestamp(), ByteUtil.EMPTY_BYTE_ARRAY)));
}
m = new Put(row, currentValueKV.getTimestamp());
- m.getFamilyMap().putAll(append.getFamilyMap());
+ m.getFamilyCellMap().putAll(append.getFamilyCellMap());
break;
case DROP_SEQUENCE:
- m = new Delete(row, clientTimestamp, null);
+ m = new Delete(row, clientTimestamp);
break;
case CREATE_SEQUENCE:
m = new Put(row, clientTimestamp);
- m.getFamilyMap().putAll(append.getFamilyMap());
+ m.getFamilyCellMap().putAll(append.getFamilyCellMap());
break;
}
if (!hadClientTimestamp) {
- for (List<KeyValue> kvs : m.getFamilyMap().values()) {
- for (KeyValue kv : kvs) {
- kv.updateLatestStamp(clientTimestampBuf);
+ for (List<Cell> kvs : m.getFamilyCellMap().values()) {
+ for (Cell kv : kvs) {
+ ((KeyValue)kv).updateLatestStamp(clientTimestampBuf);
}
}
}
- @SuppressWarnings("unchecked")
- Pair<Mutation,Integer>[] mutations = new Pair[1];
- mutations[0] = new Pair<Mutation,Integer>(m, lid);
+ Mutation[] mutations = new Mutation[]{m};
region.batchMutate(mutations);
long serverTimestamp = MetaDataUtil.getClientTimeStamp(m);
// Return result with single KeyValue. The only piece of information
// the client cares about is the timestamp, which is the timestamp of
// when the mutation was actually performed (useful in the case of .
- return new Result(Collections.singletonList(KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, serverTimestamp, SUCCESS_VALUE)));
+ return Result.create(Collections.singletonList(
+ (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, serverTimestamp, SUCCESS_VALUE)));
} finally {
- region.releaseRowLock(lid);
+ region.releaseRowLocks(locks);
}
} catch (Throwable t) {
ServerUtil.throwIOException("Increment of sequence " + Bytes.toStringBinary(row), t);
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/53f7d3ce/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
index e624c9a..9331547 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
@@ -19,18 +19,29 @@
*/
package org.apache.phoenix.coprocessor;
+import java.io.IOException;
import java.sql.SQLException;
-import org.apache.hadoop.hbase.coprocessor.BaseEndpointCoprocessor;
+import org.apache.hadoop.hbase.Coprocessor;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-
import org.apache.hadoop.hbase.index.util.ImmutableBytesPtr;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.phoenix.cache.GlobalCache;
import org.apache.phoenix.cache.TenantCache;
+import org.apache.phoenix.coprocessor.ServerCachingProtocol.ServerCacheFactory;
+import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.AddServerCacheRequest;
+import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.AddServerCacheResponse;
+import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.RemoveServerCacheRequest;
+import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.RemoveServerCacheResponse;
+import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.ServerCachingService;
+import org.apache.phoenix.protobuf.ProtobufUtil;
-
-
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.Service;
/**
*
@@ -39,19 +50,74 @@ import org.apache.phoenix.cache.TenantCache;
*
* @since 0.1
*/
-public class ServerCachingEndpointImpl extends BaseEndpointCoprocessor implements ServerCachingProtocol {
+public class ServerCachingEndpointImpl extends ServerCachingService implements CoprocessorService,
+ Coprocessor {
+
+ private RegionCoprocessorEnvironment env;
- @Override
- public boolean addServerCache(byte[] tenantId, byte[] cacheId, ImmutableBytesWritable cachePtr, ServerCacheFactory cacheFactory) throws SQLException {
- TenantCache tenantCache = GlobalCache.getTenantCache((RegionCoprocessorEnvironment)this.getEnvironment(), tenantId == null ? null : new ImmutableBytesPtr(tenantId));
- tenantCache.addServerCache(new ImmutableBytesPtr(cacheId), cachePtr, cacheFactory);
- return true;
+ @Override
+ public void addServerCache(RpcController controller, AddServerCacheRequest request,
+ RpcCallback<AddServerCacheResponse> done) {
+ ImmutableBytesPtr tenantId = null;
+ if (request.hasTenantId()) {
+ tenantId = new ImmutableBytesPtr(request.getTenantId().toByteArray());
}
+ TenantCache tenantCache = GlobalCache.getTenantCache(this.env, tenantId);
+ ImmutableBytesWritable cachePtr =
+ org.apache.phoenix.protobuf.ProtobufUtil
+ .toImmutableBytesWritable(request.getCachePtr());
- @Override
- public boolean removeServerCache(byte[] tenantId, byte[] cacheId) throws SQLException {
- TenantCache tenantCache = GlobalCache.getTenantCache((RegionCoprocessorEnvironment)this.getEnvironment(), tenantId == null ? null : new ImmutableBytesPtr(tenantId));
- tenantCache.removeServerCache(new ImmutableBytesPtr(cacheId));
- return true;
+ try {
+ @SuppressWarnings("unchecked")
+ Class<ServerCacheFactory> serverCacheFactoryClass =
+ (Class<ServerCacheFactory>) Class.forName(request.getCacheFactory().getClassName());
+ ServerCacheFactory cacheFactory = serverCacheFactoryClass.newInstance();
+ tenantCache.addServerCache(new ImmutableBytesPtr(request.getCacheId().toByteArray()),
+ cachePtr, cacheFactory);
+ } catch (Throwable e) {
+ ProtobufUtil.setControllerException(controller, new IOException(e));
}
+ AddServerCacheResponse.Builder responseBuilder = AddServerCacheResponse.newBuilder();
+ responseBuilder.setReturn(true);
+ AddServerCacheResponse result = responseBuilder.build();
+ done.run(result);
+ }
+
+ @Override
+ public void removeServerCache(RpcController controller, RemoveServerCacheRequest request,
+ RpcCallback<RemoveServerCacheResponse> done) {
+ ImmutableBytesPtr tenantId = null;
+ if (request.hasTenantId()) {
+ tenantId = new ImmutableBytesPtr(request.getTenantId().toByteArray());
+ }
+ TenantCache tenantCache = GlobalCache.getTenantCache(this.env, tenantId);
+ try {
+ tenantCache.removeServerCache(new ImmutableBytesPtr(request.getCacheId().toByteArray()));
+ } catch (SQLException e) {
+ ProtobufUtil.setControllerException(controller, new IOException(e));
+ }
+ RemoveServerCacheResponse.Builder responseBuilder = RemoveServerCacheResponse.newBuilder();
+ responseBuilder.setReturn(true);
+ RemoveServerCacheResponse result = responseBuilder.build();
+ done.run(result);
+ }
+
+ @Override
+ public void start(CoprocessorEnvironment env) throws IOException {
+ if (env instanceof RegionCoprocessorEnvironment) {
+ this.env = (RegionCoprocessorEnvironment) env;
+ } else {
+ throw new CoprocessorException("Must be loaded on a table region!");
+ }
+ }
+
+ @Override
+ public void stop(CoprocessorEnvironment arg0) throws IOException {
+ // nothing to do
+ }
+
+ @Override
+ public Service getService() {
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/53f7d3ce/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
index 3c6123d..3115216 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
@@ -23,7 +23,6 @@ import java.io.Closeable;
import java.sql.SQLException;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
import org.apache.hadoop.io.Writable;
import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
@@ -37,7 +36,7 @@ import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
*
* @since 0.1
*/
-public interface ServerCachingProtocol extends CoprocessorProtocol {
+public interface ServerCachingProtocol {
public static interface ServerCacheFactory extends Writable {
public Closeable newCache(ImmutableBytesWritable cachePtr, MemoryChunk chunk) throws SQLException;
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/53f7d3ce/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index 9207588..42de847 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -36,6 +36,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Set;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
@@ -47,10 +48,8 @@ import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.regionserver.HRegion;
-import org.apache.hadoop.hbase.regionserver.MultiVersionConsistencyControl;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.io.WritableUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,6 +57,7 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.phoenix.client.KeyValueBuilder;
+import org.apache.phoenix.coprocessor.generated.PTableProtos;
import org.apache.phoenix.exception.ValueTypeIncompatibleException;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.ExpressionType;
@@ -90,7 +90,8 @@ import org.apache.phoenix.util.SchemaUtil;
* @since 0.1
*/
public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver {
- private static final Logger logger = LoggerFactory.getLogger(UngroupedAggregateRegionObserver.class);
+ private static final Logger logger = LoggerFactory.getLogger(UngroupedAggregateRegionObserver.class);
+
// TODO: move all constants into a single class
public static final String UNGROUPED_AGG = "UngroupedAgg";
public static final String DELETE_AGG = "DeleteAgg";
@@ -107,16 +108,16 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
this.kvBuilder = KeyValueBuilder.get(e.getHBaseVersion());
}
- private static void commitBatch(HRegion region, List<Pair<Mutation,Integer>> mutations, byte[] indexUUID) throws IOException {
- if (indexUUID != null) {
- for (Pair<Mutation,Integer> pair : mutations) {
- pair.getFirst().setAttribute(PhoenixIndexCodec.INDEX_UUID, indexUUID);
- }
- }
- @SuppressWarnings("unchecked")
- Pair<Mutation,Integer>[] mutationArray = new Pair[mutations.size()];
- // TODO: should we use the one that is all or none?
- region.batchMutate(mutations.toArray(mutationArray));
+ private static void commitBatch(HRegion region, List<Mutation> mutations, byte[] indexUUID) throws IOException {
+ if (indexUUID != null) {
+ for (Mutation m : mutations) {
+ m.setAttribute(PhoenixIndexCodec.INDEX_UUID, indexUUID);
+ }
+ }
+ @SuppressWarnings("unchecked")
+ Mutation[] mutationArray = new Mutation[mutations.size()];
+ // TODO: should we use the one that is all or none?
+ region.batchMutate(mutations.toArray(mutationArray));
}
public static void serializeIntoScan(Scan scan) {
@@ -168,7 +169,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
int batchSize = 0;
long ts = scan.getTimeRange().getMax();
HRegion region = c.getEnvironment().getRegion();
- List<Pair<Mutation,Integer>> mutations = Collections.emptyList();
+ List<Mutation> mutations = Collections.emptyList();
if (isDelete || isUpsert || (deleteCQ != null && deleteCF != null) || emptyCF != null) {
// TODO: size better
mutations = Lists.newArrayListWithExpectedSize(1024);
@@ -184,17 +185,16 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
logger.info("Starting ungrouped coprocessor scan " + scan);
}
long rowCount = 0;
- MultiVersionConsistencyControl.setThreadReadPoint(innerScanner.getMvccReadPoint());
region.startRegionOperation();
try {
do {
- List<KeyValue> results = new ArrayList<KeyValue>();
+ List<Cell> results = new ArrayList<Cell>();
// Results are potentially returned even when the return value of s.next is false
// since this is an indication of whether or not there are more values after the
// ones returned
- hasMore = innerScanner.nextRaw(results, null);
+ hasMore = innerScanner.nextRaw(results);
if (!results.isEmpty()) {
- rowCount++;
+ rowCount++;
result.setKeyValues(results);
try {
if (isDelete) {
@@ -202,8 +202,10 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
// FIXME: the version of the Delete constructor without the lock args was introduced
// in 0.94.4, thus if we try to use it here we can no longer use the 0.94.2 version
// of the client.
- Delete delete = new Delete(results.get(0).getRow(),ts,null);
- mutations.add(new Pair<Mutation,Integer>(delete,null));
+ Cell firstKV = results.get(0);
+ Delete delete = new Delete(firstKV.getRowArray(), firstKV.getRowOffset(),
+ firstKV.getRowLength(),ts);
+ mutations.add(delete);
} else if (isUpsert) {
Arrays.fill(values, null);
int i = 0;
@@ -246,15 +248,16 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
}
}
for (Mutation mutation : row.toRowMutations()) {
- mutations.add(new Pair<Mutation,Integer>(mutation,null));
+ mutations.add(mutation);
}
} else if (deleteCF != null && deleteCQ != null) {
// No need to search for delete column, since we project only it
// if no empty key value is being set
if (emptyCF == null || result.getValue(deleteCF, deleteCQ) != null) {
- Delete delete = new Delete(results.get(0).getRow());
+ Delete delete = new Delete(results.get(0).getRowArray(), results.get(0).getRowOffset(),
+ results.get(0).getRowLength());
delete.deleteColumns(deleteCF, deleteCQ, ts);
- mutations.add(new Pair<Mutation,Integer>(delete,null));
+ mutations.add(delete);
}
}
if (emptyCF != null) {
@@ -267,18 +270,18 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
* We insert one empty key value per row per timestamp.
*/
Set<Long> timeStamps = Sets.newHashSetWithExpectedSize(results.size());
- for (KeyValue kv : results) {
+ for (Cell kv : results) {
long kvts = kv.getTimestamp();
if (!timeStamps.contains(kvts)) {
- Put put = new Put(kv.getRow());
+ Put put = new Put(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength());
put.add(emptyCF, QueryConstants.EMPTY_COLUMN_BYTES, kvts, ByteUtil.EMPTY_BYTE_ARRAY);
- mutations.add(new Pair<Mutation,Integer>(put,null));
+ mutations.add(put);
}
}
}
// Commit in batches based on UPSERT_BATCH_SIZE_ATTRIB in config
if (!mutations.isEmpty() && batchSize > 0 && mutations.size() % batchSize == 0) {
- commitBatch(region,mutations, indexUUID);
+ commitBatch(region, mutations, indexUUID);
mutations.clear();
}
} catch (ConstraintViolationException e) {
@@ -330,32 +333,28 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
}
@Override
- public boolean next(List<KeyValue> results) throws IOException {
+ public boolean next(List<Cell> results) throws IOException {
if (done) return false;
done = true;
results.add(aggKeyValue);
return false;
}
+
+ @Override
+ public long getMaxResultSize() {
+ return scan.getMaxResultSize();
+ }
};
return scanner;
}
private static PTable deserializeTable(byte[] b) {
- ByteArrayInputStream stream = new ByteArrayInputStream(b);
try {
- DataInputStream input = new DataInputStream(stream);
- PTable table = new PTableImpl();
- table.readFields(input);
- return table;
+ PTableProtos.PTable ptableProto = PTableProtos.PTable.parseFrom(b);
+ return PTableImpl.createFromProto(ptableProto);
} catch (IOException e) {
throw new RuntimeException(e);
- } finally {
- try {
- stream.close();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
+ }
}
private static List<Expression> deserializeExpressions(byte[] b) {
@@ -383,20 +382,8 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
}
public static byte[] serialize(PTable projectedTable) {
- ByteArrayOutputStream stream = new ByteArrayOutputStream();
- try {
- DataOutputStream output = new DataOutputStream(stream);
- projectedTable.write(output);
- return stream.toByteArray();
- } catch (IOException e) {
- throw new RuntimeException(e);
- } finally {
- try {
- stream.close();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
+ PTableProtos.PTable ptableProto = PTableImpl.toProto(projectedTable);
+ return ptableProto.toByteArray();
}
public static byte[] serialize(List<Expression> selectExpressions) {