| /* |
| * 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.coprocessor; |
| |
| import java.io.IOException; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Map; |
| |
| 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; |
| import org.apache.hadoop.hbase.client.Delete; |
| import org.apache.hadoop.hbase.client.Get; |
| import org.apache.hadoop.hbase.client.Increment; |
| import org.apache.hadoop.hbase.client.Mutation; |
| import org.apache.hadoop.hbase.client.Put; |
| import org.apache.hadoop.hbase.client.Result; |
| import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver; |
| 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.Region; |
| import org.apache.hadoop.hbase.regionserver.Region.RowLock; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.phoenix.exception.SQLExceptionCode; |
| import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; |
| import org.apache.phoenix.query.QueryConstants; |
| import org.apache.phoenix.schema.Sequence; |
| import org.apache.phoenix.schema.SortOrder; |
| import org.apache.phoenix.schema.types.PBoolean; |
| import org.apache.phoenix.schema.types.PDataType; |
| import org.apache.phoenix.schema.types.PInteger; |
| import org.apache.phoenix.schema.types.PLong; |
| import org.apache.phoenix.util.ByteUtil; |
| import org.apache.phoenix.util.EnvironmentEdgeManager; |
| import org.apache.phoenix.util.KeyValueUtil; |
| import org.apache.phoenix.util.MetaDataUtil; |
| import org.apache.phoenix.util.SequenceUtil; |
| import org.apache.phoenix.util.ServerUtil; |
| |
| import com.google.common.collect.Lists; |
| |
| /** |
| * |
| * Region observer coprocessor for sequence operations: |
| * 1) For creating a sequence, as checkAndPut does not allow us to scope the |
| * Get done for the check with a TimeRange. |
| * 2) For incrementing a sequence, as increment does not a) allow us to set the |
| * timestamp of the key value being incremented and b) recognize when the key |
| * value being incremented does not exist |
| * 3) For deleting a sequence, as checkAndDelete does not allow us to scope |
| * the Get done for the check with a TimeRange. |
| * |
| * |
| * @since 3.0.0 |
| */ |
| public class SequenceRegionObserver extends BaseRegionObserver { |
| public static final String OPERATION_ATTRIB = "SEQUENCE_OPERATION"; |
| public static final String MAX_TIMERANGE_ATTRIB = "MAX_TIMERANGE"; |
| public static final String CURRENT_VALUE_ATTRIB = "CURRENT_VALUE"; |
| public static final String NUM_TO_ALLOCATE = "NUM_TO_ALLOCATE"; |
| private static final byte[] SUCCESS_VALUE = PInteger.INSTANCE.toBytes(Integer.valueOf(Sequence.SUCCESS)); |
| |
| private static Result getErrorResult(byte[] row, long timestamp, int errorCode) { |
| byte[] errorCodeBuf = new byte[PInteger.INSTANCE.getByteSize()]; |
| PInteger.INSTANCE.getCodec().encodeInt(errorCode, errorCodeBuf, 0); |
| return Result.create(Collections.singletonList( |
| (Cell)KeyValueUtil.newKeyValue(row, |
| PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, |
| QueryConstants.EMPTY_COLUMN_BYTES, timestamp, errorCodeBuf))); |
| } |
| |
| private static void acquireLock(Region region, byte[] key, List<RowLock> locks) |
| throws IOException { |
| RowLock rowLock = region.getRowLock(key, false); |
| 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 |
| * implementation (HBASE-10254): |
| * 1) Lack of recognition and identification of when the key value to increment doesn't exist |
| * 2) Lack of the ability to set the timestamp of the updated key value. |
| * Works the same as existing region.increment(), except assumes there is a single column to |
| * increment and uses Phoenix LONG encoding. |
| * |
| * @since 3.0.0 |
| */ |
| @Override |
| public Result preIncrement(final ObserverContext<RegionCoprocessorEnvironment> e, |
| final Increment increment) throws IOException { |
| RegionCoprocessorEnvironment env = e.getEnvironment(); |
| // We need to set this to prevent region.increment from being called |
| e.bypass(); |
| e.complete(); |
| Region region = env.getRegion(); |
| byte[] row = increment.getRow(); |
| List<RowLock> locks = Lists.newArrayList(); |
| TimeRange tr = increment.getTimeRange(); |
| region.startRegionOperation(); |
| try { |
| acquireLock(region, row, locks); |
| try { |
| long maxTimestamp = tr.getMax(); |
| boolean validateOnly = true; |
| Get get = new Get(row); |
| get.setTimeRange(tr.getMin(), tr.getMax()); |
| for (Map.Entry<byte[], List<Cell>> entry : increment.getFamilyCellMap().entrySet()) { |
| byte[] cf = entry.getKey(); |
| for (Cell cq : entry.getValue()) { |
| long value = Bytes.toLong(cq.getValueArray(), cq.getValueOffset()); |
| get.addColumn(cf, CellUtil.cloneQualifier(cq)); |
| long cellTimestamp = cq.getTimestamp(); |
| // Workaround HBASE-15698 by using the lowest of the timestamps found |
| // on the Increment or any of its Cells. |
| if (cellTimestamp > 0 && cellTimestamp < maxTimestamp) { |
| maxTimestamp = cellTimestamp; |
| get.setTimeRange(MetaDataProtocol.MIN_TABLE_TIMESTAMP, maxTimestamp); |
| } |
| validateOnly &= (Sequence.ValueOp.VALIDATE_SEQUENCE.ordinal() == value); |
| } |
| } |
| Result result = region.get(get); |
| if (result.isEmpty()) { |
| return getErrorResult(row, maxTimestamp, SQLExceptionCode.SEQUENCE_UNDEFINED.getErrorCode()); |
| } |
| |
| |
| KeyValue currentValueKV = Sequence.getCurrentValueKV(result); |
| KeyValue incrementByKV = Sequence.getIncrementByKV(result); |
| KeyValue cacheSizeKV = Sequence.getCacheSizeKV(result); |
| |
| long currentValue = PLong.INSTANCE.getCodec().decodeLong(currentValueKV.getValueArray(), currentValueKV.getValueOffset(), SortOrder.getDefault()); |
| long incrementBy = PLong.INSTANCE.getCodec().decodeLong(incrementByKV.getValueArray(), incrementByKV.getValueOffset(), SortOrder.getDefault()); |
| long cacheSize = PLong.INSTANCE.getCodec().decodeLong(cacheSizeKV.getValueArray(), cacheSizeKV.getValueOffset(), SortOrder.getDefault()); |
| |
| // Hold timestamp constant for sequences, so that clients always only see the latest |
| // value regardless of when they connect. |
| long timestamp = currentValueKV.getTimestamp(); |
| Put put = new Put(row, timestamp); |
| |
| int numIncrementKVs = increment.getFamilyCellMap().get(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES).size(); |
| // creates the list of KeyValues used for the Result that will be returned |
| List<Cell> cells = Sequence.getCells(result, numIncrementKVs); |
| |
| //if client is 3.0/4.0 preserve the old behavior (older clients won't have newer columns present in the increment) |
| if (numIncrementKVs != Sequence.NUM_SEQUENCE_KEY_VALUES) { |
| currentValue += incrementBy * cacheSize; |
| // Hold timestamp constant for sequences, so that clients always only see the latest value |
| // regardless of when they connect. |
| KeyValue newCurrentValueKV = createKeyValue(row, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, currentValue, timestamp); |
| put.add(newCurrentValueKV); |
| Sequence.replaceCurrentValueKV(cells, newCurrentValueKV); |
| } |
| else { |
| KeyValue cycleKV = Sequence.getCycleKV(result); |
| KeyValue limitReachedKV = Sequence.getLimitReachedKV(result); |
| KeyValue minValueKV = Sequence.getMinValueKV(result); |
| KeyValue maxValueKV = Sequence.getMaxValueKV(result); |
| |
| boolean increasingSeq = incrementBy > 0 ? true : false; |
| |
| // if the minValue, maxValue, cycle and limitReached is null this sequence has been upgraded from |
| // a lower version. Set minValue, maxValue, cycle and limitReached to Long.MIN_VALUE, Long.MAX_VALUE, true and false |
| // respectively in order to maintain existing behavior and also update the KeyValues on the server |
| boolean limitReached; |
| if (limitReachedKV == null) { |
| limitReached = false; |
| KeyValue newLimitReachedKV = createKeyValue(row, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, limitReached, timestamp); |
| put.add(newLimitReachedKV); |
| Sequence.replaceLimitReachedKV(cells, newLimitReachedKV); |
| } |
| else { |
| limitReached = (Boolean) PBoolean.INSTANCE.toObject(limitReachedKV.getValueArray(), |
| limitReachedKV.getValueOffset(), limitReachedKV.getValueLength()); |
| } |
| long minValue; |
| if (minValueKV == null) { |
| minValue = Long.MIN_VALUE; |
| KeyValue newMinValueKV = createKeyValue(row, PhoenixDatabaseMetaData.MIN_VALUE_BYTES, minValue, timestamp); |
| put.add(newMinValueKV); |
| Sequence.replaceMinValueKV(cells, newMinValueKV); |
| } |
| else { |
| minValue = PLong.INSTANCE.getCodec().decodeLong(minValueKV.getValueArray(), |
| minValueKV.getValueOffset(), SortOrder.getDefault()); |
| } |
| long maxValue; |
| if (maxValueKV == null) { |
| maxValue = Long.MAX_VALUE; |
| KeyValue newMaxValueKV = createKeyValue(row, PhoenixDatabaseMetaData.MAX_VALUE_BYTES, maxValue, timestamp); |
| put.add(newMaxValueKV); |
| Sequence.replaceMaxValueKV(cells, newMaxValueKV); |
| } |
| else { |
| maxValue = PLong.INSTANCE.getCodec().decodeLong(maxValueKV.getValueArray(), |
| maxValueKV.getValueOffset(), SortOrder.getDefault()); |
| } |
| boolean cycle; |
| if (cycleKV == null) { |
| cycle = false; |
| KeyValue newCycleKV = createKeyValue(row, PhoenixDatabaseMetaData.CYCLE_FLAG_BYTES, cycle, timestamp); |
| put.add(newCycleKV); |
| Sequence.replaceCycleValueKV(cells, newCycleKV); |
| } |
| else { |
| cycle = (Boolean) PBoolean.INSTANCE.toObject(cycleKV.getValueArray(), |
| cycleKV.getValueOffset(), cycleKV.getValueLength()); |
| } |
| |
| long numSlotsToAllocate = calculateNumSlotsToAllocate(increment); |
| |
| // We don't support Bulk Allocations on sequences that have the CYCLE flag set to true |
| if (cycle && !SequenceUtil.isCycleAllowed(numSlotsToAllocate)) { |
| return getErrorResult(row, maxTimestamp, SQLExceptionCode.NUM_SEQ_TO_ALLOCATE_NOT_SUPPORTED.getErrorCode()); |
| } |
| |
| // Bulk Allocations are expressed by NEXT <n> VALUES FOR |
| if (SequenceUtil.isBulkAllocation(numSlotsToAllocate)) { |
| if (SequenceUtil.checkIfLimitReached(currentValue, minValue, maxValue, incrementBy, cacheSize, numSlotsToAllocate)) { |
| // If we try to allocate more slots than the limit we return an error. |
| // Allocating sequence values in bulk should be an all or nothing operation. |
| // If the operation succeeds clients are guaranteed that they have reserved |
| // all the slots requested. |
| return getErrorResult(row, maxTimestamp, SequenceUtil.getLimitReachedErrorCode(increasingSeq).getErrorCode()); |
| } |
| } |
| |
| if (validateOnly) { |
| return result; |
| } |
| |
| // return if we have run out of sequence values |
| if (limitReached) { |
| if (cycle) { |
| // reset currentValue of the Sequence row to minValue/maxValue |
| currentValue = increasingSeq ? minValue : maxValue; |
| } |
| else { |
| return getErrorResult(row, maxTimestamp, SequenceUtil.getLimitReachedErrorCode(increasingSeq).getErrorCode()); |
| } |
| } |
| |
| // check if the limit was reached |
| limitReached = SequenceUtil.checkIfLimitReached(currentValue, minValue, maxValue, incrementBy, cacheSize, numSlotsToAllocate); |
| |
| // update currentValue |
| currentValue += incrementBy * (SequenceUtil.isBulkAllocation(numSlotsToAllocate) ? numSlotsToAllocate : cacheSize); |
| // update the currentValue of the Result row |
| KeyValue newCurrentValueKV = createKeyValue(row, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, currentValue, timestamp); |
| Sequence.replaceCurrentValueKV(cells, newCurrentValueKV); |
| put.add(newCurrentValueKV); |
| // set the LIMIT_REACHED column to true, so that no new values can be used |
| KeyValue newLimitReachedKV = createKeyValue(row, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, limitReached, timestamp); |
| put.add(newLimitReachedKV); |
| } |
| // update the KeyValues on the server |
| Mutation[] mutations = new Mutation[]{put}; |
| region.batchMutate(mutations, HConstants.NO_NONCE, HConstants.NO_NONCE); |
| // return a Result with the updated KeyValues |
| return Result.create(cells); |
| } finally { |
| region.releaseRowLocks(locks); |
| } |
| } catch (Throwable t) { |
| ServerUtil.throwIOException("Increment of sequence " + Bytes.toStringBinary(row), t); |
| return null; // Impossible |
| } finally { |
| region.closeRegionOperation(); |
| } |
| } |
| |
| |
| /** |
| * Creates a new KeyValue for a long value |
| * |
| * @param key |
| * key used while creating KeyValue |
| * @param cqBytes |
| * column qualifier of KeyValue |
| * @return return the KeyValue that was created |
| */ |
| KeyValue createKeyValue(byte[] key, byte[] cqBytes, long value, long timestamp) { |
| byte[] valueBuffer = new byte[PLong.INSTANCE.getByteSize()]; |
| PLong.INSTANCE.getCodec().encodeLong(value, valueBuffer, 0); |
| return KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, cqBytes, timestamp, valueBuffer); |
| } |
| |
| /** |
| * Creates a new KeyValue for a boolean value and adds it to the given put |
| * |
| * @param key |
| * key used while creating KeyValue |
| * @param cqBytes |
| * column qualifier of KeyValue |
| * @return return the KeyValue that was created |
| */ |
| private KeyValue createKeyValue(byte[] key, byte[] cqBytes, boolean value, long timestamp) throws IOException { |
| // create new key value for put |
| return KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, cqBytes, |
| timestamp, value ? PDataType.TRUE_BYTES : PDataType.FALSE_BYTES); |
| } |
| |
| /** |
| * Override the preAppend for checkAndPut and checkAndDelete, as we need the ability to |
| * a) set the TimeRange for the Get being done and |
| * b) return something back to the client to indicate success/failure |
| */ |
| @SuppressWarnings("deprecation") |
| @Override |
| public Result preAppend(final ObserverContext<RegionCoprocessorEnvironment> e, |
| final Append append) throws IOException { |
| byte[] opBuf = append.getAttribute(OPERATION_ATTRIB); |
| if (opBuf == null) { |
| return null; |
| } |
| Sequence.MetaOp op = Sequence.MetaOp.values()[opBuf[0]]; |
| Cell keyValue = append.getFamilyCellMap().values().iterator().next().iterator().next(); |
| |
| long clientTimestamp = HConstants.LATEST_TIMESTAMP; |
| long minGetTimestamp = MetaDataProtocol.MIN_TABLE_TIMESTAMP; |
| long maxGetTimestamp = HConstants.LATEST_TIMESTAMP; |
| boolean hadClientTimestamp; |
| byte[] clientTimestampBuf = null; |
| if (op == Sequence.MetaOp.RETURN_SEQUENCE) { |
| // When returning sequences, this allows us to send the expected timestamp |
| // of the sequence to make sure we don't reset any other sequence |
| hadClientTimestamp = true; |
| clientTimestamp = minGetTimestamp = keyValue.getTimestamp(); |
| maxGetTimestamp = minGetTimestamp + 1; |
| } else { |
| clientTimestampBuf = append.getAttribute(MAX_TIMERANGE_ATTRIB); |
| if (clientTimestampBuf != null) { |
| clientTimestamp = maxGetTimestamp = Bytes.toLong(clientTimestampBuf); |
| } |
| hadClientTimestamp = (clientTimestamp != HConstants.LATEST_TIMESTAMP); |
| if (hadClientTimestamp) { |
| // Prevent race condition of creating two sequences at the same timestamp |
| // by looking for a sequence at or after the timestamp at which it'll be |
| // created. |
| if (op == Sequence.MetaOp.CREATE_SEQUENCE) { |
| maxGetTimestamp = clientTimestamp + 1; |
| } |
| } else { |
| clientTimestamp = EnvironmentEdgeManager.currentTimeMillis(); |
| maxGetTimestamp = clientTimestamp + 1; |
| clientTimestampBuf = Bytes.toBytes(clientTimestamp); |
| } |
| } |
| |
| RegionCoprocessorEnvironment env = e.getEnvironment(); |
| // We need to set this to prevent region.append from being called |
| e.bypass(); |
| e.complete(); |
| Region region = env.getRegion(); |
| byte[] row = append.getRow(); |
| List<RowLock> locks = Lists.newArrayList(); |
| region.startRegionOperation(); |
| try { |
| acquireLock(region, row, locks); |
| try { |
| byte[] family = CellUtil.cloneFamily(keyValue); |
| byte[] qualifier = CellUtil.cloneQualifier(keyValue); |
| |
| Get get = new Get(row); |
| get.setTimeRange(minGetTimestamp, maxGetTimestamp); |
| get.addColumn(family, qualifier); |
| Result result = region.get(get); |
| if (result.isEmpty()) { |
| if (op == Sequence.MetaOp.DROP_SEQUENCE || op == Sequence.MetaOp.RETURN_SEQUENCE) { |
| return getErrorResult(row, clientTimestamp, SQLExceptionCode.SEQUENCE_UNDEFINED.getErrorCode()); |
| } |
| } else { |
| if (op == Sequence.MetaOp.CREATE_SEQUENCE) { |
| return getErrorResult(row, clientTimestamp, SQLExceptionCode.SEQUENCE_ALREADY_EXIST.getErrorCode()); |
| } |
| } |
| Mutation m = null; |
| switch (op) { |
| case RETURN_SEQUENCE: |
| KeyValue currentValueKV = result.raw()[0]; |
| long expectedValue = PLong.INSTANCE.getCodec().decodeLong(append.getAttribute(CURRENT_VALUE_ATTRIB), 0, SortOrder.getDefault()); |
| long value = PLong.INSTANCE.getCodec().decodeLong(currentValueKV.getValueArray(), |
| currentValueKV.getValueOffset(), SortOrder.getDefault()); |
| // Timestamp should match exactly, or we may have the wrong sequence |
| if (expectedValue != value || currentValueKV.getTimestamp() != clientTimestamp) { |
| return Result.create(Collections.singletonList( |
| (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, |
| QueryConstants.EMPTY_COLUMN_BYTES, currentValueKV.getTimestamp(), ByteUtil.EMPTY_BYTE_ARRAY))); |
| } |
| m = new Put(row, currentValueKV.getTimestamp()); |
| m.getFamilyCellMap().putAll(append.getFamilyCellMap()); |
| break; |
| case DROP_SEQUENCE: |
| m = new Delete(row, clientTimestamp); |
| break; |
| case CREATE_SEQUENCE: |
| m = new Put(row, clientTimestamp); |
| m.getFamilyCellMap().putAll(append.getFamilyCellMap()); |
| break; |
| } |
| if (!hadClientTimestamp) { |
| for (List<Cell> kvs : m.getFamilyCellMap().values()) { |
| for (Cell kv : kvs) { |
| ((KeyValue)kv).updateLatestStamp(clientTimestampBuf); |
| } |
| } |
| } |
| Mutation[] mutations = new Mutation[]{m}; |
| region.batchMutate(mutations, HConstants.NO_NONCE, HConstants.NO_NONCE); |
| 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 Result.create(Collections.singletonList( |
| (Cell)KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, serverTimestamp, SUCCESS_VALUE))); |
| } finally { |
| region.releaseRowLocks(locks); |
| } |
| } catch (Throwable t) { |
| ServerUtil.throwIOException("Increment of sequence " + Bytes.toStringBinary(row), t); |
| return null; // Impossible |
| } finally { |
| region.closeRegionOperation(); |
| } |
| } |
| |
| /** |
| * Determines whether a request for incrementing the sequence was a bulk allocation and if so |
| * what the number of slots to allocate is. This is triggered by the NEXT <n> VALUES FOR expression. |
| * For backwards compatibility with older clients, we default the value to 1 which preserves |
| * existing behavior when invoking NEXT VALUE FOR. |
| */ |
| private long calculateNumSlotsToAllocate(final Increment increment) { |
| long numToAllocate = 1; |
| byte[] numToAllocateBytes = increment.getAttribute(SequenceRegionObserver.NUM_TO_ALLOCATE); |
| if (numToAllocateBytes != null) { |
| numToAllocate = Bytes.toLong(numToAllocateBytes); |
| } |
| return numToAllocate; |
| } |
| |
| } |
