@Override
public MetaDataMutationResult updateIndexState(final List<Mutation> tableMetaData, String parentTableName) throws SQLException {
    byte[][] rowKeyMetadata = new byte[3][];
    SchemaUtil.getVarChars(tableMetaData.get(0).getRow(), rowKeyMetadata);
    byte[] tableKey = SchemaUtil.getTableKey(ByteUtil.EMPTY_BYTE_ARRAY, rowKeyMetadata[PhoenixDatabaseMetaData.SCHEMA_NAME_INDEX], rowKeyMetadata[PhoenixDatabaseMetaData.TABLE_NAME_INDEX]);
    return metaDataCoprocessorExec(tableKey,
            new Batch.Call<MetaDataService, MetaDataResponse>() {
                @Override
                public MetaDataResponse call(MetaDataService instance) throws IOException {
                    ServerRpcController controller = new ServerRpcController();
                    BlockingRpcCallback<MetaDataResponse> rpcCallback =
                            new BlockingRpcCallback<MetaDataResponse>();
                    UpdateIndexStateRequest.Builder builder = UpdateIndexStateRequest.newBuilder();
                    for (Mutation m : tableMetaData) {
                        MutationProto mp = ProtobufUtil.toProto(m);
                        builder.addTableMetadataMutations(mp.toByteString());
                    }
                    instance.updateIndexState(controller, builder.build(), rpcCallback);
                    if(controller.getFailedOn() != null) {
                        throw controller.getFailedOn();
                    }
                    return rpcCallback.get();
                }
            });
}
 

/**
 * Each ByteString entry is a byte array serialized from MutationProto instance
 * @param mutations
 * @throws IOException
 */
private static List<Mutation> getMutations(List<ByteString> mutations)
        throws IOException {
    List<Mutation> result = new ArrayList<Mutation>();
    for (ByteString mutation : mutations) {
        MutationProto mProto = MutationProto.parseFrom(mutation);
        result.add(org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(mProto));
    }
    return result;
}
 

public static MutationProto toProto(Mutation mutation) throws IOException {
    MutationType type;
    if (mutation instanceof Put) {
        type = MutationType.PUT;
    } else if (mutation instanceof Delete) {
        type = MutationType.DELETE;
    } else {
        throw new IllegalArgumentException("Only Put and Delete are supported");
    }
    return org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(type, mutation);
}
 

private byte[] getMutationBytes() {
    try {
        MutationProto m = toMutationProto(this.mutation);
        return m.toByteArray();
    } catch (IOException e) {
        throw new IllegalArgumentException("Failed to get bytes for mutation!", e);
    }
}
 

/**
 * This method shouldn't be used - you should use {@link KeyValueCodec#readKeyValue(DataInput)} instead. Its the
 * complement to {@link #writeData(DataOutput)}.
 */
@SuppressWarnings("javadoc")
public void readFields(DataInput in) throws IOException {
    this.indexTableName = new ImmutableBytesPtr(Bytes.readByteArray(in));
    byte[] mutationData = Bytes.readByteArray(in);
    MutationProto mProto = MutationProto.parseFrom(mutationData);
    this.mutation = org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(mProto);
    this.hashCode = calcHashCode(indexTableName, mutation);
}
 

protected MutationProto toMutationProto(Mutation mutation)  throws IOException {
    MutationProto m = null;
    if(mutation instanceof Put){
        m = org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(MutationType.PUT, 
            mutation);
    } else if(mutation instanceof Delete) {
        m = org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(MutationType.DELETE, 
            mutation);
    } else {
        throw new IOException("Put/Delete mutations only supported");
    }
    return m;
}
 

/**
 * Each ByteString entry is a byte array serialized from MutationProto instance
 * @param mutations
 * @throws IOException
 */
private static List<Mutation> getMutations(List<ByteString> mutations)
        throws IOException {
    List<Mutation> result = new ArrayList<Mutation>();
    for (ByteString mutation : mutations) {
        MutationProto mProto = MutationProto.parseFrom(mutation);
        result.add(org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(mProto));
    }
    return result;
}
 

public static MutationProto toProto(Mutation mutation) throws IOException {
    MutationType type;
    if (mutation instanceof Put) {
        type = MutationType.PUT;
    } else if (mutation instanceof Delete) {
        type = MutationType.DELETE;
    } else {
        throw new IllegalArgumentException("Only Put and Delete are supported");
    }
    return org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(type, mutation);
}
 

private byte[] getMutationBytes() {
    try {
        MutationProto m = toMutationProto(this.mutation);
        return m.toByteArray();
    } catch (IOException e) {
        throw new IllegalArgumentException("Failed to get bytes for mutation!", e);
    }
}
 

protected MutationProto toMutationProto(Mutation mutation)  throws IOException {
    MutationProto m = null;
    if(mutation instanceof Put){
        m = org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(MutationType.PUT, 
            mutation);
    } else if(mutation instanceof Delete) {
        m = org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(MutationType.DELETE, 
            mutation);
    } else {
        throw new IOException("Put/Delete mutations only supported");
    }
    return m;
}
 

public static MetaDataMutationResult updateIndexState(byte[] indexTableKey, long minTimeStamp,
        Table metaTable, PIndexState newState) throws Throwable {
    // Mimic the Put that gets generated by the client on an update of the index state
    Put put = new Put(indexTableKey);
    put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES,
            newState.getSerializedBytes());
    put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES,
            PLong.INSTANCE.toBytes(minTimeStamp));
    put.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.ASYNC_REBUILD_TIMESTAMP_BYTES,
            PLong.INSTANCE.toBytes(0));
    final List<Mutation> tableMetadata = Collections.<Mutation> singletonList(put);

    final Map<byte[], MetaDataResponse> results = metaTable.coprocessorService(MetaDataService.class, indexTableKey,
            indexTableKey, new Batch.Call<MetaDataService, MetaDataResponse>() {
                @Override
                public MetaDataResponse call(MetaDataService instance) throws IOException {
                    ServerRpcController controller = new ServerRpcController();
                    BlockingRpcCallback<MetaDataResponse> rpcCallback = new BlockingRpcCallback<MetaDataResponse>();
                    UpdateIndexStateRequest.Builder builder = UpdateIndexStateRequest.newBuilder();
                    for (Mutation m : tableMetadata) {
                        MutationProto mp = ProtobufUtil.toProto(m);
                        builder.addTableMetadataMutations(mp.toByteString());
                    }
                    builder.setClientVersion(VersionUtil.encodeVersion(PHOENIX_MAJOR_VERSION, PHOENIX_MINOR_VERSION, PHOENIX_PATCH_NUMBER));
                    instance.updateIndexState(controller, builder.build(), rpcCallback);
                    if (controller.getFailedOn() != null) { throw controller.getFailedOn(); }
                    return rpcCallback.get();
                }
            });
    if (results.isEmpty()) { throw new IOException("Didn't get expected result size"); }
    MetaDataResponse tmpResponse = results.values().iterator().next();
    return MetaDataMutationResult.constructFromProto(tmpResponse);
}
 

@Override
public void encode(Mutation mutation, OutputStream outStream) throws IOException {
  MutationType type = getType(mutation);
  MutationProto proto = ProtobufUtil.toMutation(type, mutation);
  proto.writeDelimitedTo(outStream);
}
 

@Override
public Mutation decode(InputStream inStream) throws IOException {
  return ProtobufUtil.toMutation(MutationProto.parseDelimitedFrom(inStream));
}
 

@Override
public MetaDataMutationResult dropTable(final List<Mutation> tableMetaData, final PTableType tableType, final boolean cascade) throws SQLException {
    byte[][] rowKeyMetadata = new byte[3][];
    SchemaUtil.getVarChars(tableMetaData.get(0).getRow(), rowKeyMetadata);
    byte[] tenantIdBytes = rowKeyMetadata[PhoenixDatabaseMetaData.TENANT_ID_INDEX];
    byte[] schemaBytes = rowKeyMetadata[PhoenixDatabaseMetaData.SCHEMA_NAME_INDEX];
    byte[] tableBytes = rowKeyMetadata[PhoenixDatabaseMetaData.TABLE_NAME_INDEX];
    byte[] tableKey = SchemaUtil.getTableKey(tenantIdBytes == null ? ByteUtil.EMPTY_BYTE_ARRAY : tenantIdBytes, schemaBytes, tableBytes);
    final MetaDataMutationResult result =  metaDataCoprocessorExec(tableKey,
            new Batch.Call<MetaDataService, MetaDataResponse>() {
                @Override
                public MetaDataResponse call(MetaDataService instance) throws IOException {
                    ServerRpcController controller = new ServerRpcController();
                    BlockingRpcCallback<MetaDataResponse> rpcCallback =
                            new BlockingRpcCallback<MetaDataResponse>();
                    DropTableRequest.Builder builder = DropTableRequest.newBuilder();
                    for (Mutation m : tableMetaData) {
                        MutationProto mp = ProtobufUtil.toProto(m);
                        builder.addTableMetadataMutations(mp.toByteString());
                    }
                    builder.setTableType(tableType.getSerializedValue());
                    builder.setCascade(cascade);

                    instance.dropTable(controller, builder.build(), rpcCallback);
                    if(controller.getFailedOn() != null) {
                        throw controller.getFailedOn();
                    }
                    return rpcCallback.get();
                }
            });

    final MutationCode code = result.getMutationCode();
    switch(code) {
    case TABLE_ALREADY_EXISTS:
        ReadOnlyProps props = this.getProps();
        boolean dropMetadata = props.getBoolean(DROP_METADATA_ATTRIB, DEFAULT_DROP_METADATA);
        if (dropMetadata) {
            dropTables(result.getTableNamesToDelete());
        }
        invalidateTables(result.getTableNamesToDelete());
        if (tableType == PTableType.TABLE) {
            byte[] physicalName = SchemaUtil.getTableNameAsBytes(schemaBytes, tableBytes);
            long timestamp = MetaDataUtil.getClientTimeStamp(tableMetaData);
            ensureViewIndexTableDropped(physicalName, timestamp);
            ensureLocalIndexTableDropped(physicalName, timestamp);
            tableStatsCache.invalidate(new ImmutableBytesPtr(physicalName));
        }
        break;
    default:
        break;
    }
      return result;
}
 

@Override
public MetaDataMutationResult dropColumn(final List<Mutation> tableMetaData, PTableType tableType) throws SQLException {
    byte[][] rowKeyMetadata = new byte[3][];
    SchemaUtil.getVarChars(tableMetaData.get(0).getRow(), rowKeyMetadata);
    byte[] tenantIdBytes = rowKeyMetadata[PhoenixDatabaseMetaData.TENANT_ID_INDEX];
    byte[] schemaBytes = rowKeyMetadata[PhoenixDatabaseMetaData.SCHEMA_NAME_INDEX];
    byte[] tableBytes = rowKeyMetadata[PhoenixDatabaseMetaData.TABLE_NAME_INDEX];
    byte[] tableKey = SchemaUtil.getTableKey(tenantIdBytes, schemaBytes, tableBytes);
    MetaDataMutationResult result = metaDataCoprocessorExec(tableKey,
        new Batch.Call<MetaDataService, MetaDataResponse>() {
            @Override
            public MetaDataResponse call(MetaDataService instance) throws IOException {
                ServerRpcController controller = new ServerRpcController();
                BlockingRpcCallback<MetaDataResponse> rpcCallback =
                        new BlockingRpcCallback<MetaDataResponse>();
                DropColumnRequest.Builder builder = DropColumnRequest.newBuilder();
                for (Mutation m : tableMetaData) {
                    MutationProto mp = ProtobufUtil.toProto(m);
                    builder.addTableMetadataMutations(mp.toByteString());
                }
                instance.dropColumn(controller, builder.build(), rpcCallback);
                if(controller.getFailedOn() != null) {
                    throw controller.getFailedOn();
                }
                return rpcCallback.get();
            }
        });
    final MutationCode code = result.getMutationCode();
    switch(code) {
    case TABLE_ALREADY_EXISTS:
        final ReadOnlyProps props = this.getProps();
        final boolean dropMetadata = props.getBoolean(DROP_METADATA_ATTRIB, DEFAULT_DROP_METADATA);
        if (dropMetadata) {
            dropTables(result.getTableNamesToDelete());
        }
        invalidateTables(result.getTableNamesToDelete());
        break;
    default:
        break;
    }
    return result;

}
 

/**
 * Internal write the underlying data for the entry - this does not do any special prefixing. Writing should be done
 * via {@link KeyValueCodec#write(DataOutput, KeyValue)} to ensure consistent reading/writing of
 * {@link IndexedKeyValue}s.
 * 
 * @param out
 *            to write data to. Does not close or flush the passed object.
 * @throws IOException
 *             if there is a problem writing the underlying data
 */
void writeData(DataOutput out) throws IOException {
    Bytes.writeByteArray(out, this.indexTableName.get());
    MutationProto m = toMutationProto(this.mutation);
    Bytes.writeByteArray(out, m.toByteArray());
}
 

/**
 * Internal write the underlying data for the entry - this does not do any special prefixing.
 * Writing should be done via {@link KeyValueCodec#write(DataOutput, KeyValue)} to ensure
 * consistent reading/writing of {@link IndexedKeyValue}s.
 * 
 * @param out
 *            to write data to. Does not close or flush the passed object.
 * @throws IOException
 *             if there is a problem writing the underlying data
 */
void writeData(DataOutput out) throws IOException {
    Bytes.writeByteArray(out, this.indexTableName.get());
    MutationProto m = toMutationProto(this.mutation);
    Bytes.writeByteArray(out, m.toByteArray());
}