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/03/10 07:20:47 UTC
[30/50] merge master to 4.0 branch
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 71a287d,27c4c5a..dff2a11
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@@ -50,13 -81,12 +82,15 @@@ import org.apache.hadoop.hbase.client.H
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Result;
+ import org.apache.hadoop.hbase.client.ResultScanner;
+ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
- import org.apache.hadoop.hbase.index.Indexer;
- import org.apache.hadoop.hbase.index.covered.CoveredColumnsIndexBuilder;
+ import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+ import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
+import org.apache.hadoop.hbase.ipc.ServerRpcController;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.VersionInfo;
@@@ -100,17 -118,23 +136,25 @@@ import org.apache.phoenix.schema.EmptyS
import org.apache.phoenix.schema.MetaDataSplitPolicy;
import org.apache.phoenix.schema.NewerTableAlreadyExistsException;
import org.apache.phoenix.schema.PColumn;
+ import org.apache.phoenix.schema.PColumnFamily;
+ import org.apache.phoenix.schema.PDataType;
import org.apache.phoenix.schema.PMetaData;
import org.apache.phoenix.schema.PMetaDataImpl;
+ import org.apache.phoenix.schema.PName;
+ import org.apache.phoenix.schema.PNameFactory;
import org.apache.phoenix.schema.PTable;
+ import org.apache.phoenix.schema.PTable.LinkType;
+ import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.ReadOnlyTableException;
+ import org.apache.phoenix.schema.SaltingUtil;
import org.apache.phoenix.schema.Sequence;
import org.apache.phoenix.schema.SequenceKey;
+ import org.apache.phoenix.schema.TableAlreadyExistsException;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.util.ByteUtil;
+import org.apache.phoenix.util.ConfigUtil;
+import org.apache.phoenix.util.JDBCUtil;
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.ReadOnlyProps;
@@@ -122,7 -148,7 +168,8 @@@ import com.google.common.collect.Iterat
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+ import com.google.common.io.Closeables;
+import com.google.protobuf.HBaseZeroCopyByteString;
public class ConnectionQueryServicesImpl extends DelegateQueryServices implements ConnectionQueryServices {
private static final Logger logger = LoggerFactory.getLogger(ConnectionQueryServicesImpl.class);
@@@ -165,9 -203,17 +224,19 @@@
// Without making a copy of the configuration we cons up, we lose some of our properties
// on the server side during testing.
this.config = HBaseConfiguration.create(config);
+ // set replication required parameter
+ ConfigUtil.setReplicationConfigIfAbsent(this.config);
this.props = new ReadOnlyProps(this.config.iterator());
+ try {
+ this.connection = HBaseFactoryProvider.getHConnectionFactory().createConnection(this.config);
- } catch (ZooKeeperConnectionException e) {
++ } catch (IOException e) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION)
+ .setRootCause(e).build().buildException();
+ }
+ if (this.connection.isClosed()) { // TODO: why the heck doesn't this throw above?
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION).build().buildException();
+ }
+ this.latestMetaData = newEmptyMetaData();
// TODO: should we track connection wide memory usage or just org-wide usage?
// If connection-wide, create a MemoryManager here, otherwise just use the one from the delegate
this.childServices = new ConcurrentHashMap<ImmutableBytesWritable,ConnectionQueryServices>(INITIAL_CHILD_SERVICES_CAPACITY);
@@@ -200,10 -246,13 +269,19 @@@
@Override
public HTableDescriptor getTableDescriptor(byte[] tableName) throws SQLException {
+ HTableInterface htable = getTable(tableName);
try {
- return getTable(tableName).getTableDescriptor();
+ return htable.getTableDescriptor();
} catch (IOException e) {
++ if(e instanceof org.apache.hadoop.hbase.TableNotFoundException ||
++ e.getCause() instanceof org.apache.hadoop.hbase.TableNotFoundException) {
++ byte[][] schemaAndTableName = new byte[2][];
++ SchemaUtil.getVarChars(tableName, schemaAndTableName);
++ throw new TableNotFoundException(Bytes.toString(schemaAndTableName[0]), Bytes.toString(schemaAndTableName[1]));
++ }
throw new RuntimeException(e);
+ } finally {
+ Closeables.closeQuietly(htable);
}
}
@@@ -439,12 -496,14 +526,15 @@@
Object value = entry.getValue();
hcd.setValue(key, value == null ? null : value.toString());
}
- hcd.setKeepDeletedCells(true);
}
+ private static final String OLD_PACKAGE = "com.salesforce.";
+ private static final String NEW_PACKAGE = "org.apache.";
+
private HTableDescriptor generateTableDescriptor(byte[] tableName, HTableDescriptor existingDesc, PTableType tableType, Map<String,Object> tableProps, List<Pair<byte[],Map<String,Object>>> families, byte[][] splits) throws SQLException {
+ String defaultFamilyName = (String)tableProps.remove(PhoenixDatabaseMetaData.DEFAULT_COLUMN_FAMILY_NAME);
- HTableDescriptor descriptor = (existingDesc != null) ? new HTableDescriptor(existingDesc) : new HTableDescriptor(tableName);
+ HTableDescriptor descriptor = (existingDesc != null) ? new HTableDescriptor(existingDesc) :
+ new HTableDescriptor(TableName.valueOf(tableName));
for (Entry<String,Object> entry : tableProps.entrySet()) {
String key = entry.getKey();
Object value = entry.getValue();
@@@ -727,21 -813,18 +844,22 @@@
serverMap.add(entry);
}
}
- HTableInterface ht = this.getTable(PhoenixDatabaseMetaData.TYPE_TABLE_NAME_BYTES);
- final TreeMap<byte[],Long> results = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
- connection.processExecs(MetaDataProtocol.class, regionKeys,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, this.getDelegate().getExecutor(), new Batch.Call<MetaDataProtocol,Long>() {
- @Override
- public Long call(MetaDataProtocol instance) throws IOException {
- return instance.getVersion();
- }
- },
- new Batch.Callback<Long>(){
++
++ HTableInterface ht = this.getTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
+ final Map<byte[], Long> results =
+ ht.coprocessorService(MetaDataService.class, null, null, new Batch.Call<MetaDataService,Long>() {
@Override
- public void update(byte[] region, byte[] row, Long value) {
- results.put(region, value);
+ public Long call(MetaDataService instance) throws IOException {
+ ServerRpcController controller = new ServerRpcController();
+ BlockingRpcCallback<GetVersionResponse> rpcCallback =
+ new BlockingRpcCallback<GetVersionResponse>();
+ GetVersionRequest.Builder builder = GetVersionRequest.newBuilder();
+
+ instance.getVersion(controller, builder.build(), rpcCallback);
+ if(controller.getFailedOn() != null) {
+ throw controller.getFailedOn();
+ }
+ return rpcCallback.get().getVersion();
}
});
for (Map.Entry<byte[],Long> result : results.entrySet()) {
@@@ -779,20 -861,21 +897,20 @@@
try {
boolean retried = false;
while (true) {
- HRegionLocation regionLocation = retried ? connection.relocateRegion(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, tableKey) : connection.locateRegion(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, tableKey);
- List<byte[]> regionKeys = Collections.singletonList(regionLocation.getRegionInfo().getStartKey());
- final Map<byte[],MetaDataMutationResult> results = Maps.newHashMapWithExpectedSize(1);
- connection.processExecs(MetaDataProtocol.class, regionKeys,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, this.getDelegate().getExecutor(), callable,
- new Batch.Callback<MetaDataMutationResult>(){
- @Override
- public void update(byte[] region, byte[] row, MetaDataMutationResult value) {
- results.put(region, value);
- }
- });
+ if (retried) {
+ connection.relocateRegion(
- TableName.valueOf(PhoenixDatabaseMetaData.TYPE_TABLE_NAME_BYTES),
++ TableName.valueOf(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES),
+ tableKey);
+ }
+
- HTableInterface ht = this.getTable(PhoenixDatabaseMetaData.TYPE_TABLE_NAME_BYTES);
++ HTableInterface ht = this.getTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
+ final Map<byte[], MetaDataResponse> results =
+ ht.coprocessorService(MetaDataService.class, tableKey, tableKey, callable);
+
assert(results.size() == 1);
- MetaDataMutationResult result = results.values().iterator().next();
- if (result.getMutationCode() == MutationCode.TABLE_NOT_IN_REGION) {
- if (retried) return result;
+ MetaDataResponse result = results.values().iterator().next();
+ if (result.getReturnCode() == MetaDataProtos.MutationCode.TABLE_NOT_IN_REGION) {
+ if (retried) return MetaDataMutationResult.constructFromProto(result);
retried = true;
continue;
}
@@@ -847,31 -991,17 +1038,31 @@@
}
@Override
- public MetaDataMutationResult getTable(byte[] tenantId, final byte[] schemaBytes, final byte[] tableBytes,
+ public MetaDataMutationResult getTable(final PName tenantId, final byte[] schemaBytes, final byte[] tableBytes,
final long tableTimestamp, final long clientTimestamp) throws SQLException {
- final byte[] nonNullTenantId = tenantId == null ? ByteUtil.EMPTY_BYTE_ARRAY : tenantId;
- byte[] tableKey = SchemaUtil.getTableKey(nonNullTenantId, schemaBytes, tableBytes);
+ final byte[] tenantIdBytes = tenantId == null ? ByteUtil.EMPTY_BYTE_ARRAY : tenantId.getBytes();
+ byte[] tableKey = SchemaUtil.getTableKey(tenantIdBytes, schemaBytes, tableBytes);
return metaDataCoprocessorExec(tableKey,
- new Batch.Call<MetaDataProtocol, MetaDataMutationResult>() {
- @Override
- public MetaDataMutationResult call(MetaDataProtocol instance) throws IOException {
- return instance.getTable(tenantIdBytes, schemaBytes, tableBytes, tableTimestamp, clientTimestamp);
- }
- });
+ new Batch.Call<MetaDataService, MetaDataResponse>() {
+ @Override
+ public MetaDataResponse call(MetaDataService instance) throws IOException {
+ ServerRpcController controller = new ServerRpcController();
+ BlockingRpcCallback<MetaDataResponse> rpcCallback =
+ new BlockingRpcCallback<MetaDataResponse>();
+ GetTableRequest.Builder builder = GetTableRequest.newBuilder();
- builder.setTenantId(HBaseZeroCopyByteString.wrap(nonNullTenantId));
++ builder.setTenantId(HBaseZeroCopyByteString.wrap(tenantIdBytes));
+ builder.setSchemaName(HBaseZeroCopyByteString.wrap(schemaBytes));
+ builder.setTableName(HBaseZeroCopyByteString.wrap(tableBytes));
+ builder.setTableTimestamp(tableTimestamp);
+ builder.setClientTimestamp(clientTimestamp);
+
+ instance.getTable(controller, builder.build(), rpcCallback);
+ if(controller.getFailedOn() != null) {
+ throw controller.getFailedOn();
+ }
+ return rpcCallback.get();
+ }
+ });
}
@Override
@@@ -978,26 -1155,34 +1230,47 @@@
if (tableMetaData.size() == 1 && tableMetaData.get(0).isEmpty()) {
return null;
}
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
MetaDataMutationResult result = metaDataCoprocessorExec(tableKey,
- new Batch.Call<MetaDataProtocol, MetaDataMutationResult>() {
+ new Batch.Call<MetaDataService, MetaDataResponse>() {
@Override
- public MetaDataMutationResult call(MetaDataProtocol instance) throws IOException {
- return instance.addColumn(tableMetaData);
+ public MetaDataResponse call(MetaDataService instance) throws IOException {
+ ServerRpcController controller = new ServerRpcController();
+ BlockingRpcCallback<MetaDataResponse> rpcCallback =
+ new BlockingRpcCallback<MetaDataResponse>();
+ AddColumnRequest.Builder builder = AddColumnRequest.newBuilder();
+ for (Mutation m : tableMetaData) {
+ MutationProto mp = ProtobufUtil.toProto(m);
+ builder.addTableMetadataMutations(mp.toByteString());
+ }
+
+ instance.addColumn(controller, builder.build(), rpcCallback);
+ if(controller.getFailedOn() != null) {
+ throw controller.getFailedOn();
+ }
+ return rpcCallback.get();
}
});
+
+ if (result.getMutationCode() == MutationCode.COLUMN_NOT_FOUND) { // Success
+ // Flush the table if transitioning DISABLE_WAL from TRUE to FALSE
+ if (Boolean.FALSE.equals(PDataType.BOOLEAN.toObject(
+ MetaDataUtil.getMutationKVByteValue(m,PhoenixDatabaseMetaData.DISABLE_WAL_BYTES, kvBuilder, ptr)))) {
+ flushTable(table.getPhysicalName().getBytes());
+ }
+
+ if (tableType == PTableType.TABLE) {
+ // If we're changing MULTI_TENANT to true or false, create or drop the view index table
+ if (MetaDataUtil.getMutationKeyValue(m, PhoenixDatabaseMetaData.MULTI_TENANT_BYTES, kvBuilder, ptr)){
+ long timestamp = MetaDataUtil.getClientTimeStamp(m);
+ if (Boolean.TRUE.equals(PDataType.BOOLEAN.toObject(ptr.get(), ptr.getOffset(), ptr.getLength()))) {
+ this.ensureViewIndexTableCreated(table, timestamp);
+ } else {
+ this.ensureViewIndexTableDropped(table.getPhysicalName().getBytes(), timestamp);
+ }
+ }
+ }
+ }
return result;
}
@@@ -1074,19 -1247,9 +1347,19 @@@
@Override
public void init(String url, Properties props) throws SQLException {
+ try {
+ this.connection = HConnectionManager.createConnection(this.config);
+ } catch (IOException ioe) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION)
+ .setRootCause(ioe).build().buildException();
+ }
+ if (this.connection.isClosed()) { // TODO: why the heck doesn't this throw above?
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ESTABLISH_CONNECTION).build().buildException();
+ }
+
props = new Properties(props);
props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP));
- PhoenixConnection metaConnection = new PhoenixConnection(this, url, props, PMetaDataImpl.EMPTY_META_DATA);
+ PhoenixConnection metaConnection = new PhoenixConnection(this, url, props, newEmptyMetaData());
SQLException sqlE = null;
try {
try {
@@@ -1141,21 -1308,14 +1418,21 @@@
protected void clearCache() throws SQLException {
try {
SQLException sqlE = null;
- HTableInterface htable = this.getTable(PhoenixDatabaseMetaData.TYPE_TABLE_NAME_BYTES);
+ HTableInterface htable = this.getTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
try {
- htable.coprocessorExec(MetaDataProtocol.class, HConstants.EMPTY_START_ROW,
- HConstants.EMPTY_END_ROW, new Batch.Call<MetaDataProtocol, Void>() {
+ htable.coprocessorService(MetaDataService.class, HConstants.EMPTY_START_ROW,
+ HConstants.EMPTY_END_ROW, new Batch.Call<MetaDataService, ClearCacheResponse>() {
@Override
- public Void call(MetaDataProtocol instance) throws IOException {
- instance.clearCache();
- return null;
+ public ClearCacheResponse call(MetaDataService instance) throws IOException {
+ ServerRpcController controller = new ServerRpcController();
+ BlockingRpcCallback<ClearCacheResponse> rpcCallback =
+ new BlockingRpcCallback<ClearCacheResponse>();
+ ClearCacheRequest.Builder builder = ClearCacheRequest.newBuilder();
+ instance.clearCache(controller, builder.build(), rpcCallback);
+ if(controller.getFailedOn() != null) {
+ throw controller.getFailedOn();
+ }
+ return rpcCallback.get();
}
});
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index c514cb2,ea53149..369abcb
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@@ -247,8 -252,9 +256,9 @@@ public class ConnectionlessQueryService
public MetaDataMutationResult updateIndexState(List<Mutation> tableMetadata, String parentTableName) throws SQLException {
byte[][] rowKeyMetadata = new byte[3][];
SchemaUtil.getVarChars(tableMetadata.get(0).getRow(), rowKeyMetadata);
- KeyValue newKV = tableMetadata.get(0).getFamilyMap().get(TABLE_FAMILY_BYTES).get(0);
- PIndexState newState = PIndexState.fromSerializedValue(newKV.getBuffer()[newKV.getValueOffset()]);
+ Cell newKV = tableMetadata.get(0).getFamilyCellMap().get(TABLE_FAMILY_BYTES).get(0);
+ PIndexState newState = PIndexState.fromSerializedValue(newKV.getValueArray()[newKV.getValueOffset()]);
+ byte[] tenantIdBytes = rowKeyMetadata[PhoenixDatabaseMetaData.TENANT_ID_INDEX];
String schemaName = Bytes.toString(rowKeyMetadata[PhoenixDatabaseMetaData.SCHEMA_NAME_INDEX]);
String indexName = Bytes.toString(rowKeyMetadata[PhoenixDatabaseMetaData.TABLE_NAME_INDEX]);
String indexTableName = SchemaUtil.getTableName(schemaName, indexName);
@@@ -361,4 -368,4 +372,4 @@@
public boolean supportsFeature(Feature feature) {
return false;
}
--}
++}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/query/HConnectionFactory.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateColumn.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 45745d6,7e74289..2fd88e6
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@@ -1097,12 -1238,12 +1238,12 @@@ public class MetaDataClient
long clientTimeStamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
List<Mutation> tableMetaData = Lists.newArrayListWithExpectedSize(2);
@SuppressWarnings("deprecation") // FIXME: Remove when unintentionally deprecated method is fixed (HBASE-7870).
- Delete tableDelete = new Delete(key, clientTimeStamp, null);
+ Delete tableDelete = new Delete(key, clientTimeStamp);
tableMetaData.add(tableDelete);
if (parentTableName != null) {
- byte[] linkKey = MetaDataUtil.getParentLinkKey(tenantId, schemaName, parentTableName, tableName);
+ byte[] linkKey = MetaDataUtil.getParentLinkKey(tenantIdStr, schemaName, parentTableName, tableName);
@SuppressWarnings("deprecation") // FIXME: Remove when unintentionally deprecated method is fixed (HBASE-7870).
- Delete linkDelete = new Delete(linkKey, clientTimeStamp, null);
+ Delete linkDelete = new Delete(linkKey, clientTimeStamp);
tableMetaData.add(linkDelete);
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/PColumn.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/PColumnImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/PColumnImpl.java
index a16c188,d702507..8e4dcd6
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PColumnImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PColumnImpl.java
@@@ -19,12 -17,23 +17,14 @@@
*/
package org.apache.phoenix.schema;
- import com.google.protobuf.HBaseZeroCopyByteString;
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.WritableUtils;
+import org.apache.phoenix.coprocessor.generated.PTableProtos;
import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.util.ByteUtil;
+ import org.apache.phoenix.util.SizedUtil;
+
import com.google.common.base.Preconditions;
-
++import com.google.protobuf.HBaseZeroCopyByteString;
public class PColumnImpl implements PColumn {
- private static final Integer NO_MAXLENGTH = Integer.MIN_VALUE;
- private static final Integer NO_SCALE = Integer.MIN_VALUE;
-
private PName name;
private PName familyName;
private PDataType dataType;
@@@ -164,56 -225,8 +175,68 @@@
return arraySize;
}
+ @Override
+ public byte[] getViewConstant() {
+ return viewConstant;
+ }
-}
++
+ /**
+ * Create a PColumn instance from PBed PColumn instance
+ * @param column
+ */
+ public static PColumn createFromProto(PTableProtos.PColumn column) {
+ byte[] columnNameBytes = column.getColumnNameBytes().toByteArray();
+ PName columnName = PNameFactory.newName(columnNameBytes);
+ PName familyName = null;
+ if (column.hasFamilyNameBytes()) {
+ familyName = PNameFactory.newName(column.getFamilyNameBytes().toByteArray());
+ }
+ PDataType dataType = PDataType.fromSqlTypeName(column.getDataType());
+ Integer maxLength = null;
+ if (column.hasMaxLength()) {
+ maxLength = column.getMaxLength();
+ }
+ Integer scale = null;
+ if (column.hasScale()) {
+ scale = column.getScale();
+ }
+ boolean nullable = column.getNullable();
+ int position = column.getPosition();
- ColumnModifier columnModifier = ColumnModifier.fromSystemValue(column.getColumnModifier());
++ SortOrder sortOrder = SortOrder.fromSystemValue(column.getSortOrder());
+ Integer arraySize = null;
+ if(column.hasArraySize()){
+ arraySize = column.getArraySize();
+ }
++ byte[] viewConstant = null;
++ if (column.hasViewConstant()) {
++ viewConstant = column.getViewConstant().toByteArray();
++ }
+ return new PColumnImpl(columnName, familyName, dataType, maxLength, scale, nullable, position,
- columnModifier, arraySize);
++ sortOrder, arraySize, viewConstant);
+ }
+
+ public static PTableProtos.PColumn toProto(PColumn column) {
+ PTableProtos.PColumn.Builder builder = PTableProtos.PColumn.newBuilder();
+ builder.setColumnNameBytes(HBaseZeroCopyByteString.wrap(column.getName().getBytes()));
+ if (column.getFamilyName() != null) {
+ builder.setFamilyNameBytes(HBaseZeroCopyByteString.wrap(column.getFamilyName().getBytes()));
+ }
+ builder.setDataType(column.getDataType().getSqlTypeName());
+ if (column.getMaxLength() != null) {
+ builder.setMaxLength(column.getMaxLength());
+ }
+ if (column.getScale() != null) {
+ builder.setScale(column.getScale());
+ }
+ builder.setNullable(column.isNullable());
+ builder.setPosition(column.getPosition());
- builder.setColumnModifier(ColumnModifier.toSystemValue(column.getColumnModifier()));
++ builder.setSortOrder(column.getSortOrder().getSystemValue());
+ if(column.getArraySize() != null){
+ builder.setArraySize(column.getArraySize());
+ }
++ if(column.getViewConstant() != null){
++ builder.setViewConstant(HBaseZeroCopyByteString.wrap(column.getViewConstant()));
++ }
+ return builder.build();
+ }
- }
++}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
index 3827c6b,37301ae..6d1b53e
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
@@@ -32,21 -32,18 +30,21 @@@ import java.util.HashMap
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
- import org.apache.hadoop.hbase.index.util.ImmutableBytesPtr;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.client.KeyValueBuilder;
+import org.apache.phoenix.coprocessor.generated.PTableProtos;
+ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.protobuf.ProtobufUtil;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.RowKeySchema.RowKeySchemaBuilder;
import org.apache.phoenix.schema.stat.PTableStats;
@@@ -746,140 -918,19 +787,176 @@@ public class PTableImpl implements PTab
public boolean isWALDisabled() {
return disableWAL;
}
+
+ @Override
+ public Short getViewIndexId() {
+ return viewIndexId;
+ }
+
+ @Override
+ public PName getTenantId() {
+ return tenantId;
+ }
+
+ /**
+ * Construct a PTable instance from ProtoBuffered PTable instance
+ * @param table
+ */
+ public static PTable createFromProto(PTableProtos.PTable table) {
++ PName tenantId = null;
++ if(table.hasTenantId()){
++ tenantId = PNameFactory.newName(table.getTenantId().toByteArray());
++ }
+ PName schemaName = PNameFactory.newName(table.getSchemaNameBytes().toByteArray());
+ PName tableName = PNameFactory.newName(table.getTableNameBytes().toByteArray());
+ PTableType tableType = PTableType.values()[table.getTableType().ordinal()];
- PIndexState indexState = null;
++ PIndexState indexState = null;
+ if (table.hasIndexState()) {
+ indexState = PIndexState.fromSerializedValue(table.getIndexState());
+ }
++ Short viewIndexId = null;
++ if(table.hasViewIndexId()){
++ viewIndexId = (short)table.getViewIndexId();
++ }
+ long sequenceNumber = table.getSequenceNumber();
+ long timeStamp = table.getTimeStamp();
+ PName pkName = null;
+ if (table.hasPkNameBytes()) {
+ pkName = PNameFactory.newName(table.getPkNameBytes().toByteArray());
+ }
+ int bucketNum = table.getBucketNum();
+ List<PColumn> columns = Lists.newArrayListWithExpectedSize(table.getColumnsCount());
+ for (PTableProtos.PColumn curPColumnProto : table.getColumnsList()) {
+ columns.add(PColumnImpl.createFromProto(curPColumnProto));
+ }
+ List<PTable> indexes = Lists.newArrayListWithExpectedSize(table.getIndexesCount());
+ for (PTableProtos.PTable curPTableProto : table.getIndexesList()) {
+ indexes.add(createFromProto(curPTableProto));
+ }
+ boolean isImmutableRows = table.getIsImmutableRows();
+ Map<String, byte[][]> guidePosts = new HashMap<String, byte[][]>();
+ for (PTableProtos.PTableStats pTableStatsProto : table.getGuidePostsList()) {
+ byte[][] value = new byte[pTableStatsProto.getValuesCount()][];
+ for (int j = 0; j < pTableStatsProto.getValuesCount(); j++) {
+ value[j] = pTableStatsProto.getValues(j).toByteArray();
+ }
+ guidePosts.put(pTableStatsProto.getKey(), value);
+ }
+ PName dataTableName = null;
+ if (table.hasDataTableNameBytes()) {
+ dataTableName = PNameFactory.newName(table.getDataTableNameBytes().toByteArray());
+ }
+ PName defaultFamilyName = null;
+ if (table.hasDefaultFamilyName()) {
+ defaultFamilyName = PNameFactory.newName(table.getDefaultFamilyName().toByteArray());
+ }
+ boolean disableWAL = table.getDisableWAL();
+ boolean multiTenant = table.getMultiTenant();
+ ViewType viewType = null;
+ String viewStatement = null;
+ List<PName> physicalNames = Collections.emptyList();
+ if (tableType == PTableType.VIEW) {
- viewType = ViewType.fromSerializedValue(table.getViewType().toByteArray()[0]);
- if(table.hasViewStatement()){
- viewStatement = (String)PDataType.VARCHAR.toObject(table.getViewStatement().toByteArray());
- physicalNames = Lists.newArrayListWithExpectedSize(table.getPhysicalNamesCount());
- for(int i = 0; i < table.getPhysicalNamesCount(); i++){
- physicalNames.add(PNameFactory.newName(table.getPhysicalNames(i).toByteArray()));
- }
- }
- }
++ viewType = ViewType.fromSerializedValue(table.getViewType().toByteArray()[0]);
++ if(table.hasViewStatement()){
++ viewStatement = (String)PDataType.VARCHAR.toObject(table.getViewStatement().toByteArray());
++ }
++ }
++ if (tableType == PTableType.VIEW || viewIndexId != null) {
++ physicalNames = Lists.newArrayListWithExpectedSize(table.getPhysicalNamesCount());
++ for(int i = 0; i < table.getPhysicalNamesCount(); i++){
++ physicalNames.add(PNameFactory.newName(table.getPhysicalNames(i).toByteArray()));
++ }
++ }
++
+ PTableStats stats = new PTableStatsImpl(guidePosts);
+ try {
+ PTableImpl result = new PTableImpl();
- result.init(schemaName, tableName, tableType, indexState, timeStamp, sequenceNumber, pkName,
++ result.init(tenantId, schemaName, tableName, tableType, indexState, timeStamp, sequenceNumber, pkName,
+ (bucketNum == NO_SALTING) ? null : bucketNum, columns, stats, dataTableName,indexes, isImmutableRows,
- physicalNames, defaultFamilyName, viewStatement, disableWAL, multiTenant, viewType);
++ physicalNames, defaultFamilyName, viewStatement, disableWAL, multiTenant, viewType, viewIndexId);
+ return result;
+ } catch (SQLException e) {
+ throw new RuntimeException(e); // Impossible
+ }
+ }
+
+ public static PTableProtos.PTable toProto(PTable table) {
+ PTableProtos.PTable.Builder builder = PTableProtos.PTable.newBuilder();
++ if(table.getTenantId() != null){
++ builder.setTenantId(HBaseZeroCopyByteString.wrap(table.getTenantId().getBytes()));
++ }
+ builder.setSchemaNameBytes(HBaseZeroCopyByteString.wrap(table.getSchemaName().getBytes()));
+ builder.setTableNameBytes(HBaseZeroCopyByteString.wrap(table.getTableName().getBytes()));
+ builder.setTableType(ProtobufUtil.toPTableTypeProto(table.getType()));
- if (table.getType() == PTableType.INDEX && table.getIndexState() != null) {
- builder.setIndexState(table.getIndexState().getSerializedValue());
++ if (table.getType() == PTableType.INDEX) {
++ if(table.getIndexState() != null) {
++ builder.setIndexState(table.getIndexState().getSerializedValue());
++ }
++ if(table.getViewIndexId() != null) {
++ builder.setViewIndexId(table.getViewIndexId());
++ }
+ }
+ builder.setSequenceNumber(table.getSequenceNumber());
+ builder.setTimeStamp(table.getTimeStamp());
+ PName tmp = table.getPKName();
+ if (tmp != null) {
+ builder.setPkNameBytes(HBaseZeroCopyByteString.wrap(tmp.getBytes()));
+ }
+ Integer bucketNum = table.getBucketNum();
+ int offset = 0;
+ if(bucketNum == null){
+ builder.setBucketNum(NO_SALTING);
+ } else {
+ offset = 1;
+ builder.setBucketNum(bucketNum);
+ }
+ List<PColumn> columns = table.getColumns();
+ int columnSize = columns.size();
+ for (int i = offset; i < columnSize; i++) {
- PColumn column = columns.get(i);
- builder.addColumns(PColumnImpl.toProto(column));
++ PColumn column = columns.get(i);
++ builder.addColumns(PColumnImpl.toProto(column));
+ }
+ List<PTable> indexes = table.getIndexes();
+ for (PTable curIndex : indexes) {
+ builder.addIndexes(toProto(curIndex));
+ }
+ builder.setIsImmutableRows(table.isImmutableRows());
+
+ // build stats
+ Map<String, byte[][]> statsMap = table.getTableStats().getGuidePosts();
+ if(statsMap != null) {
+ for (Entry<String, byte[][]> entry : statsMap.entrySet()) {
+ PTableProtos.PTableStats.Builder statsBuilder = PTableProtos.PTableStats.newBuilder();
+ statsBuilder.setKey(entry.getKey());
+ for (byte[] curVal : entry.getValue()) {
+ statsBuilder.addValues(HBaseZeroCopyByteString.wrap(curVal));
+ }
+ builder.addGuidePosts(statsBuilder.build());
+ }
+ }
+ if (table.getParentName() != null) {
+ builder.setDataTableNameBytes(HBaseZeroCopyByteString.wrap(table.getParentTableName().getBytes()));
+ }
+ if (table.getDefaultFamilyName()!= null) {
+ builder.setDefaultFamilyName(HBaseZeroCopyByteString.wrap(table.getDefaultFamilyName().getBytes()));
+ }
+ builder.setDisableWAL(table.isWALDisabled());
+ builder.setMultiTenant(table.isMultiTenant());
+ if(table.getType() == PTableType.VIEW){
+ builder.setViewType(HBaseZeroCopyByteString.wrap(new byte[]{table.getViewType().getSerializedValue()}));
+ builder.setViewStatement(HBaseZeroCopyByteString.wrap(PDataType.VARCHAR.toBytes(table.getViewStatement())));
++ }
++ if(table.getType() == PTableType.VIEW || table.getViewIndexId() != null){
+ for (int i = 0; i < table.getPhysicalNames().size(); i++) {
+ builder.addPhysicalNames(HBaseZeroCopyByteString.wrap(table.getPhysicalNames().get(i).getBytes()));
+ }
+ }
+
+ return builder.build();
+ }
- }
+
+ @Override
+ public PTableKey getKey() {
+ return key;
+ }
-}
++}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
index 9f95a30,44c823f..c62edc4
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
@@@ -173,9 -194,9 +195,9 @@@ public class Sequence
// from the cache that we shouldn't have which will cause a gap in sequence values.
// In that case, we might get an error that a curr value was done on a sequence
// before a next val was. Not sure how to prevent that.
- if (result.raw().length == 1) {
- KeyValue errorKV = result.raw()[0];
- int errorCode = PDataType.INTEGER.getCodec().decodeInt(errorKV.getBuffer(), errorKV.getValueOffset(), SortOrder.getDefault());
+ if (result.rawCells().length == 1) {
+ Cell errorKV = result.rawCells()[0];
- int errorCode = PDataType.INTEGER.getCodec().decodeInt(errorKV.getValueArray(), errorKV.getValueOffset(), null);
++ int errorCode = PDataType.INTEGER.getCodec().decodeInt(errorKV.getValueArray(), errorKV.getValueOffset(), SortOrder.getDefault());
SQLExceptionCode code = SQLExceptionCode.fromErrorCode(errorCode);
// TODO: We could have the server return the timestamps of the
// delete markers and we could insert them here, but this seems
@@@ -287,9 -308,9 +309,9 @@@
KeyValue incrementByKV = getIncrementByKV(r);
KeyValue cacheSizeKV = getCacheSizeKV(r);
timestamp = currentValueKV.getTimestamp();
- nextValue = PDataType.LONG.getCodec().decodeLong(currentValueKV.getValueArray(), currentValueKV.getValueOffset(), null);
- incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getValueArray(), incrementByKV.getValueOffset(), null);
- cacheSize = PDataType.INTEGER.getCodec().decodeInt(cacheSizeKV.getValueArray(), cacheSizeKV.getValueOffset(), null);
- nextValue = PDataType.LONG.getCodec().decodeLong(currentValueKV.getBuffer(), currentValueKV.getValueOffset(), SortOrder.getDefault());
- incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getBuffer(), incrementByKV.getValueOffset(), SortOrder.getDefault());
- cacheSize = PDataType.INTEGER.getCodec().decodeInt(cacheSizeKV.getBuffer(), cacheSizeKV.getValueOffset(), SortOrder.getDefault());
++ nextValue = PDataType.LONG.getCodec().decodeLong(currentValueKV.getValueArray(), currentValueKV.getValueOffset(), SortOrder.getDefault());
++ incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getValueArray(), incrementByKV.getValueOffset(), SortOrder.getDefault());
++ cacheSize = PDataType.INTEGER.getCodec().decodeInt(cacheSizeKV.getValueArray(), cacheSizeKV.getValueOffset(), SortOrder.getDefault());
currentValue = nextValue - incrementBy * cacheSize;
}
}
@@@ -300,7 -321,7 +322,7 @@@
return false;
}
long timestamp = statusKV.getTimestamp();
- int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), null);
- int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getBuffer(), statusKV.getValueOffset(), SortOrder.getDefault());
++ int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), SortOrder.getDefault());
if (statusCode == SUCCESS) { // Success - update nextValue down to currentValue
SequenceValue value = findSequenceValue(timestamp);
if (value == null) {
@@@ -341,9 -362,9 +363,9 @@@
}
public long createSequence(Result result) throws SQLException {
- KeyValue statusKV = result.raw()[0];
+ Cell statusKV = result.rawCells()[0];
long timestamp = statusKV.getTimestamp();
- int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), null);
- int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getBuffer(), statusKV.getValueOffset(), SortOrder.getDefault());
++ int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), SortOrder.getDefault());
if (statusCode == 0) { // Success - add sequence value and return timestamp
SequenceValue value = new SequenceValue(timestamp);
insertSequenceValue(value);
@@@ -370,9 -391,9 +392,9 @@@
}
public long dropSequence(Result result) throws SQLException {
- KeyValue statusKV = result.raw()[0];
+ Cell statusKV = result.rawCells()[0];
long timestamp = statusKV.getTimestamp();
- int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), null);
- int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getBuffer(), statusKV.getValueOffset(), SortOrder.getDefault());
++ int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), SortOrder.getDefault());
SQLExceptionCode code = statusCode == 0 ? null : SQLExceptionCode.fromErrorCode(statusCode);
if (code == null) {
// Insert delete marker so that point-in-time sequences work
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStats.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/stat/PTableStatsImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/DelegateTuple.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/DelegateTuple.java
index 0000000,ec01e7e..58b1eda
mode 000000,100644..100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/DelegateTuple.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/DelegateTuple.java
@@@ -1,0 -1,64 +1,64 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ package org.apache.phoenix.schema.tuple;
+
-import org.apache.hadoop.hbase.KeyValue;
++import org.apache.hadoop.hbase.Cell;
+ import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+
+ public class DelegateTuple implements Tuple {
+ private final Tuple delegate;
+
+ public DelegateTuple(Tuple delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public int size() {
+ return delegate.size();
+ }
+
+ @Override
+ public boolean isImmutable() {
+ return delegate.isImmutable();
+ }
+
+ @Override
+ public void getKey(ImmutableBytesWritable ptr) {
+ delegate.getKey(ptr);
+ }
+
+ @Override
- public KeyValue getValue(int index) {
++ public Cell getValue(int index) {
+ return delegate.getValue(index);
+ }
+
+ @Override
- public KeyValue getValue(byte[] family, byte[] qualifier) {
++ public Cell getValue(byte[] family, byte[] qualifier) {
+ return delegate.getValue(family, qualifier);
+ }
+
+ @Override
+ public boolean getValue(byte[] family, byte[] qualifier, ImmutableBytesWritable ptr) {
+ return delegate.getValue(family, qualifier, ptr);
+ }
+
+ @Override
+ public long getSequenceValue(int index) {
+ return delegate.getSequenceValue(index);
+ }
+ }
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java
index 3395f61,3c00dfc..a1030ca
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/MultiKeyValueTuple.java
@@@ -21,18 -19,17 +19,17 @@@ package org.apache.phoenix.schema.tuple
import java.util.List;
-import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.Cell;
- import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+ import org.apache.phoenix.util.KeyValueUtil;
import com.google.common.collect.ImmutableList;
- import org.apache.phoenix.util.KeyValueUtil;
- public class MultiKeyValueTuple implements Tuple {
+ public class MultiKeyValueTuple extends BaseTuple {
- private List<KeyValue> values;
+ private List<Cell> values;
- public MultiKeyValueTuple(List<KeyValue> values) {
+ public MultiKeyValueTuple(List<Cell> values) {
setKeyValues(values);
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java
index 5b492da,2b51946..61b2a4f
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/Tuple.java
@@@ -58,9 -56,9 +56,9 @@@ public interface Tuple
* @return the KeyValue at the given index
* @throws IndexOutOfBoundsException if an invalid index is used
*/
- public KeyValue getValue(int index);
+ public Cell getValue(int index);
- /***
+ /**
* Get the KeyValue contained by the Tuple with the given family and
* qualifier name.
* @param family the column family of the KeyValue being retrieved
@@@ -68,9 -66,9 +66,9 @@@
* @return the KeyValue with the given family and qualifier name or
* null if not found.
*/
- public KeyValue getValue(byte [] family, byte [] qualifier);
+ public Cell getValue(byte [] family, byte [] qualifier);
- /***
+ /**
* Get the value byte array of the KeyValue contained by the Tuple with
* the given family and qualifier name.
* @param family the column family of the KeyValue being retrieved
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/CSVLoader.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
index a30e704,7075980..093e044
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
@@@ -24,12 -22,9 +22,9 @@@ import java.sql.SQLException
import java.util.List;
import java.util.Map;
-import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
- import org.apache.hadoop.hbase.index.ValueGetter;
- import org.apache.hadoop.hbase.index.covered.update.ColumnReference;
- import org.apache.hadoop.hbase.index.util.ImmutableBytesPtr;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.client.KeyValueBuilder;
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/KeyValueUtil.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
index 5c92aed,adf7f98..a3fd123
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
@@@ -24,7 -23,7 +23,8 @@@ import java.sql.SQLException
import java.util.Collection;
import java.util.List;
+import org.apache.hadoop.hbase.Cell;
+ import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
@@@ -152,11 -180,11 +181,11 @@@ public class MetaDataUtil
}
public static long getSequenceNumber(Mutation tableMutation) {
- List<KeyValue> kvs = tableMutation.getFamilyMap().get(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES);
+ List<Cell> kvs = tableMutation.getFamilyCellMap().get(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES);
if (kvs != null) {
- for (KeyValue kv : kvs) { // list is not ordered, so search. TODO: we could potentially assume the position
- if (Bytes.compareTo(kv.getBuffer(), kv.getQualifierOffset(), kv.getQualifierLength(), PhoenixDatabaseMetaData.TABLE_SEQ_NUM_BYTES, 0, PhoenixDatabaseMetaData.TABLE_SEQ_NUM_BYTES.length) == 0) {
- return PDataType.LONG.getCodec().decodeLong(kv.getBuffer(), kv.getValueOffset(), SortOrder.getDefault());
+ for (Cell kv : kvs) { // list is not ordered, so search. TODO: we could potentially assume the position
+ if (Bytes.compareTo(kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength(), PhoenixDatabaseMetaData.TABLE_SEQ_NUM_BYTES, 0, PhoenixDatabaseMetaData.TABLE_SEQ_NUM_BYTES.length) == 0) {
- return PDataType.LONG.getCodec().decodeLong(kv.getValueArray(), kv.getValueOffset(), null);
++ return PDataType.LONG.getCodec().decodeLong(kv.getValueArray(), kv.getValueOffset(), SortOrder.getDefault());
}
}
}
@@@ -180,21 -212,38 +213,39 @@@
return tableMetaData.get(0);
}
- public static byte[] getMutationKVByteValue(Mutation headerRow, byte[] key) {
- KeyValue kv = getMutationKeyValue(headerRow, key);
- // FIXME: byte copy
- return kv == null ? ByteUtil.EMPTY_BYTE_ARRAY : kv.getValue();
+ public static byte[] getMutationKVByteValue(Mutation headerRow, byte[] key,
+ KeyValueBuilder builder, ImmutableBytesWritable ptr) {
+ if (getMutationKeyValue(headerRow, key, builder, ptr)) {
+ return ByteUtil.copyKeyBytesIfNecessary(ptr);
+ }
+ return ByteUtil.EMPTY_BYTE_ARRAY;
}
- private static KeyValue getMutationKeyValue(Mutation headerRow, byte[] key) {
+ /**
+ * Get the mutation who's qualifier matches the passed key
+ * <p>
+ * We need to pass in an {@link ImmutableBytesPtr} to pass the result back to make life easier
+ * when dealing with a regular {@link KeyValue} vs. a {@link ClientKeyValue} as the latter doesn't
+ * support things like {@link KeyValue#getBuffer()}
+ * @param headerRow mutation to check
+ * @param key to check
+ * @param builder that created the {@link KeyValue KeyValues} in the {@link Mutation}
+ * @param ptr to update with the value of the mutation
+ * @return the value of the matching {@link KeyValue}
+ */
+ public static boolean getMutationKeyValue(Mutation headerRow, byte[] key,
+ KeyValueBuilder builder, ImmutableBytesWritable ptr) {
- List<KeyValue> kvs = headerRow.getFamilyMap().get(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES);
+ List<Cell> kvs = headerRow.getFamilyCellMap().get(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES);
if (kvs != null) {
- for (Cell kv : kvs) {
- if (Bytes.compareTo(kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength(), key, 0,
- key.length) == 0) { return org.apache.hadoop.hbase.KeyValueUtil.ensureKeyValue(kv); }
- for (KeyValue kv : kvs) {
++ for (Cell cell : kvs) {
++ KeyValue kv = org.apache.hadoop.hbase.KeyValueUtil.ensureKeyValue(cell);
+ if (builder.compareQualifier(kv, key, 0, key.length) ==0) {
+ builder.getValueAsPtr(kv, ptr);
+ return true;
+ }
}
}
- return null;
+ return false;
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/ResultUtil.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index d0ab434,b998307..7722f7c
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@@ -52,8 -49,11 +49,9 @@@ import org.apache.phoenix.schema.PTable
import org.apache.phoenix.schema.RowKeySchema;
import org.apache.phoenix.schema.RowKeySchema.RowKeySchemaBuilder;
import org.apache.phoenix.schema.SaltingUtil;
+ import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.ValueSchema.Field;
-
-
/**
*
* Static class for various schema-related utilities
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/main/java/org/apache/phoenix/util/TupleUtil.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestReadWriteKeyValuesWithCodec.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestReadWriteKeyValuesWithCodec.java
index c63d7af,f091ea7..182b5c4
--- a/phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestReadWriteKeyValuesWithCodec.java
+++ b/phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestReadWriteKeyValuesWithCodec.java
@@@ -19,10 -34,6 +37,8 @@@ import org.apache.hadoop.hbase.KeyValue
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.codec.Codec;
- import org.apache.hadoop.hbase.index.IndexTestingUtils;
- import org.apache.hadoop.hbase.index.wal.IndexedKeyValue;
+import org.apache.hadoop.hbase.io.util.LRUDictionary;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.BeforeClass;
import org.junit.Test;
@@@ -164,4 -170,4 +183,4 @@@ public class TestReadWriteKeyValuesWith
}
}
}
--}
++}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplayWithIndexWritesAndCompressedWAL.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplayWithIndexWritesAndUncompressedWALInHBase_094_9.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplayWithIndexWritesAndUncompressedWALInHBase_094_9.java
index a92eaf7,42813ed..cc7a111
--- a/phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplayWithIndexWritesAndUncompressedWALInHBase_094_9.java
+++ b/phoenix-core/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplayWithIndexWritesAndUncompressedWALInHBase_094_9.java
@@@ -2,8 -20,8 +20,9 @@@ package org.apache.hadoop.hbase.regions
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
- import org.apache.hadoop.hbase.index.util.IndexManagementUtil;
+
+ import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
+import org.apache.phoenix.util.ConfigUtil;
/**
* Do the WAL Replay test but with the WALEditCodec, rather than an {@link IndexedHLogReader}, but
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/end2end/AlterTableTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/end2end/ConnectionQueryServicesTestImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/test/java/org/apache/phoenix/end2end/ConnectionQueryServicesTestImpl.java
index 69b8752,3800261..c051feb
--- a/phoenix-core/src/test/java/org/apache/phoenix/end2end/ConnectionQueryServicesTestImpl.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/end2end/ConnectionQueryServicesTestImpl.java
@@@ -30,7 -28,7 +28,6 @@@ import org.apache.phoenix.jdbc.PhoenixE
import org.apache.phoenix.query.ConnectionQueryServicesImpl;
import org.apache.phoenix.query.QueryServices;
--
/**
*
* Implementation of ConnectionQueryServices for tests running against
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/end2end/NativeHBaseTypesTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/end2end/index/IndexTestUtil.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/test/java/org/apache/phoenix/end2end/index/IndexTestUtil.java
index 5cd9f6d,df17cf3..79cd118
--- a/phoenix-core/src/test/java/org/apache/phoenix/end2end/index/IndexTestUtil.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/end2end/index/IndexTestUtil.java
@@@ -40,9 -38,11 +38,9 @@@ import org.apache.hadoop.hbase.client.D
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
-
-import com.google.common.collect.Lists;
import org.apache.phoenix.client.KeyValueBuilder;
import org.apache.phoenix.query.QueryConstants;
- import org.apache.phoenix.schema.ColumnModifier;
+ import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.ColumnNotFoundException;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PColumnFamily;
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/filter/SkipScanFilterTest.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/test/java/org/apache/phoenix/filter/SkipScanFilterTest.java
index 886d3fc,eeaba4b..e1f8959
--- a/phoenix-core/src/test/java/org/apache/phoenix/filter/SkipScanFilterTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/filter/SkipScanFilterTest.java
@@@ -1,6 -1,23 +1,24 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
package org.apache.phoenix.filter;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexTestingUtils.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexTestingUtils.java
index 0000000,92f45b9..7fa9c8e
mode 000000,100644..100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexTestingUtils.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/IndexTestingUtils.java
@@@ -1,0 -1,94 +1,94 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ package org.apache.phoenix.hbase.index;
+
+ import static org.junit.Assert.assertEquals;
+
+ import java.io.IOException;
+ import java.util.ArrayList;
+ import java.util.List;
+
+ import org.apache.commons.logging.Log;
+ import org.apache.commons.logging.LogFactory;
+ import org.apache.hadoop.conf.Configuration;
+ import org.apache.hadoop.hbase.HConstants;
+ import org.apache.hadoop.hbase.KeyValue;
+ import org.apache.hadoop.hbase.client.HTable;
+ import org.apache.hadoop.hbase.client.Result;
+ import org.apache.hadoop.hbase.client.ResultScanner;
+ import org.apache.hadoop.hbase.client.Scan;
+ import org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec;
-import org.apache.hadoop.hbase.regionserver.wal.WALEditCodec;
++import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
+ import org.apache.hadoop.hbase.util.Bytes;
+
+
+
+ /**
+ * Utility class for testing indexing
+ */
+ public class IndexTestingUtils {
+
+ private static final Log LOG = LogFactory.getLog(IndexTestingUtils.class);
+ private static final String MASTER_INFO_PORT_KEY = "hbase.master.info.port";
+ private static final String RS_INFO_PORT_KEY = "hbase.regionserver.info.port";
+
+ private IndexTestingUtils() {
+ // private ctor for util class
+ }
+
+ public static void setupConfig(Configuration conf) {
+ conf.setInt(MASTER_INFO_PORT_KEY, -1);
+ conf.setInt(RS_INFO_PORT_KEY, -1);
+ // setup our codec, so we get proper replay/write
- conf.set(WALEditCodec.WAL_EDIT_CODEC_CLASS_KEY, IndexedWALEditCodec.class.getName());
++ conf.set(WALCellCodec.WAL_CELL_CODEC_CLASS_KEY, IndexedWALEditCodec.class.getName());
+ }
+ /**
+ * Verify the state of the index table between the given key and time ranges against the list of
+ * expected keyvalues.
+ * @throws IOException
+ */
+ @SuppressWarnings("javadoc")
+ public static void verifyIndexTableAtTimestamp(HTable index1, List<KeyValue> expected,
+ long start, long end, byte[] startKey, byte[] endKey) throws IOException {
+ LOG.debug("Scanning " + Bytes.toString(index1.getTableName()) + " between times (" + start
+ + ", " + end + "] and keys: [" + Bytes.toString(startKey) + ", " + Bytes.toString(endKey)
+ + "].");
+ Scan s = new Scan(startKey, endKey);
+ // s.setRaw(true);
+ s.setMaxVersions();
+ s.setTimeRange(start, end);
+ List<KeyValue> received = new ArrayList<KeyValue>();
+ ResultScanner scanner = index1.getScanner(s);
+ for (Result r : scanner) {
+ received.addAll(r.list());
+ LOG.debug("Received: " + r.list());
+ }
+ scanner.close();
+ assertEquals("Didn't get the expected kvs from the index table!", expected, received);
+ }
+
+ public static void verifyIndexTableAtTimestamp(HTable index1, List<KeyValue> expected, long ts,
+ byte[] startKey) throws IOException {
+ IndexTestingUtils.verifyIndexTableAtTimestamp(index1, expected, ts, startKey, HConstants.EMPTY_END_ROW);
+ }
+
+ public static void verifyIndexTableAtTimestamp(HTable index1, List<KeyValue> expected, long start,
+ byte[] startKey, byte[] endKey) throws IOException {
+ verifyIndexTableAtTimestamp(index1, expected, start, start + 1, startKey, endKey);
+ }
+ }
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/b3a330ca/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/TestFailForUnsupportedHBaseVersions.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/test/java/org/apache/phoenix/hbase/index/TestFailForUnsupportedHBaseVersions.java
index 0000000,43ec826..c9c7083
mode 000000,100644..100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/TestFailForUnsupportedHBaseVersions.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/TestFailForUnsupportedHBaseVersions.java
@@@ -1,0 -1,156 +1,159 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ package org.apache.phoenix.hbase.index;
+
+ import static org.junit.Assert.assertFalse;
+ import static org.junit.Assert.assertNotNull;
+ import static org.junit.Assert.assertNull;
+
+ import org.apache.commons.logging.Log;
+ import org.apache.commons.logging.LogFactory;
+ import org.apache.hadoop.conf.Configuration;
+ import org.apache.hadoop.hbase.HBaseConfiguration;
+ import org.apache.hadoop.hbase.HBaseTestingUtility;
+ import org.apache.hadoop.hbase.HColumnDescriptor;
+ import org.apache.hadoop.hbase.HConstants;
+ import org.apache.hadoop.hbase.HTableDescriptor;
+ import org.apache.hadoop.hbase.client.HBaseAdmin;
+ import org.apache.hadoop.hbase.regionserver.HRegionServer;
+ import org.apache.hadoop.hbase.util.Bytes;
+ import org.apache.hadoop.hbase.util.VersionInfo;
++import org.apache.phoenix.util.ConfigUtil;
+ import org.junit.Test;
+
+ import org.apache.phoenix.hbase.index.Indexer;
+ import org.apache.phoenix.hbase.index.covered.example.ColumnGroup;
+ import org.apache.phoenix.hbase.index.covered.example.CoveredColumn;
+ import org.apache.phoenix.hbase.index.covered.example.CoveredColumnIndexSpecifierBuilder;
+
+ /**
+ * Test that we correctly fail for versions of HBase that don't support current properties
+ */
+ public class TestFailForUnsupportedHBaseVersions {
+ private static final Log LOG = LogFactory.getLog(TestFailForUnsupportedHBaseVersions.class);
+
+ /**
+ * We don't support WAL Compression for HBase < 0.94.9, so we shouldn't even allow the server
+ * to start if both indexing and WAL Compression are enabled for the wrong versions.
+ */
+ @Test
+ public void testDoesNotSupportCompressedWAL() {
+ Configuration conf = HBaseConfiguration.create();
+ IndexTestingUtils.setupConfig(conf);
+ // get the current version
+ String version = VersionInfo.getVersion();
+
+ // ensure WAL Compression not enabled
+ conf.setBoolean(HConstants.ENABLE_WAL_COMPRESSION, false);
+
+ //we support all versions without WAL Compression
+ String supported = Indexer.validateVersion(version, conf);
+ assertNull(
+ "WAL Compression wasn't enabled, but version "+version+" of HBase wasn't supported! All versions should"
+ + " support writing without a compressed WAL. Message: "+supported, supported);
+
+ // enable WAL Compression
+ conf.setBoolean(HConstants.ENABLE_WAL_COMPRESSION, true);
+
+ // set the version to something we know isn't supported
+ version = "0.94.4";
+ supported = Indexer.validateVersion(version, conf);
+ assertNotNull("WAL Compression was enabled, but incorrectly marked version as supported",
+ supported);
+
+ //make sure the first version of 0.94 that supports Indexing + WAL Compression works
+ version = "0.94.9";
+ supported = Indexer.validateVersion(version, conf);
+ assertNull(
+ "WAL Compression wasn't enabled, but version "+version+" of HBase wasn't supported! Message: "+supported, supported);
+
+ //make sure we support snapshot builds too
+ version = "0.94.9-SNAPSHOT";
+ supported = Indexer.validateVersion(version, conf);
+ assertNull(
+ "WAL Compression wasn't enabled, but version "+version+" of HBase wasn't supported! Message: "+supported, supported);
+ }
+
+ /**
+ * Test that we correctly abort a RegionServer when we run tests with an unsupported HBase
+ * version. The 'completeness' of this test requires that we run the test with both a version of
+ * HBase that wouldn't be supported with WAL Compression. Currently, this is the default version
+ * (0.94.4) so just running 'mvn test' will run the full test. However, this test will not fail
+ * when running against a version of HBase with WALCompression enabled. Therefore, to fully test
+ * this functionality, we need to run the test against both a supported and an unsupported version
+ * of HBase (as long as we want to support an version of HBase that doesn't support custom WAL
+ * Codecs).
+ * @throws Exception on failure
+ */
+ @Test(timeout = 300000 /* 5 mins */)
+ public void testDoesNotStartRegionServerForUnsupportedCompressionAndVersion() throws Exception {
+ Configuration conf = HBaseConfiguration.create();
+ IndexTestingUtils.setupConfig(conf);
+ // enable WAL Compression
+ conf.setBoolean(HConstants.ENABLE_WAL_COMPRESSION, true);
+
+ // check the version to see if it isn't supported
+ String version = VersionInfo.getVersion();
+ boolean supported = false;
+ if (Indexer.validateVersion(version, conf) == null) {
+ supported = true;
+ }
+
+ // start the minicluster
+ HBaseTestingUtility util = new HBaseTestingUtility(conf);
++ // set replication required parameter
++ ConfigUtil.setReplicationConfigIfAbsent(conf);
+ util.startMiniCluster();
+
+ // setup the primary table
+ HTableDescriptor desc = new HTableDescriptor(
+ "testDoesNotStartRegionServerForUnsupportedCompressionAndVersion");
+ byte[] family = Bytes.toBytes("f");
+ desc.addFamily(new HColumnDescriptor(family));
+
+ // enable indexing to a non-existant index table
+ String indexTableName = "INDEX_TABLE";
+ ColumnGroup fam1 = new ColumnGroup(indexTableName);
+ fam1.add(new CoveredColumn(family, CoveredColumn.ALL_QUALIFIERS));
+ CoveredColumnIndexSpecifierBuilder builder = new CoveredColumnIndexSpecifierBuilder();
+ builder.addIndexGroup(fam1);
+ builder.build(desc);
+
+ // get a reference to the regionserver, so we can ensure it aborts
+ HRegionServer server = util.getMiniHBaseCluster().getRegionServer(0);
+
+ // create the primary table
+ HBaseAdmin admin = util.getHBaseAdmin();
+ if (supported) {
+ admin.createTable(desc);
+ assertFalse("Hosting regeion server failed, even the HBase version (" + version
+ + ") supports WAL Compression.", server.isAborted());
+ } else {
+ admin.createTableAsync(desc, null);
+
+ // wait for the regionserver to abort - if this doesn't occur in the timeout, assume its
+ // broken.
+ while (!server.isAborted()) {
+ LOG.debug("Waiting on regionserver to abort..");
+ }
+ }
+
+ // cleanup
+ util.shutdownMiniCluster();
+ }
+ }