| /* |
| * 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; |
| |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CACHE_SIZE_BYTES; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CYCLE_FLAG_BYTES; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INCREMENT_BY_BYTES; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MAX_VALUE_BYTES; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MIN_VALUE_BYTES; |
| import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES; |
| |
| import java.io.IOException; |
| import java.sql.SQLException; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.locks.ReentrantLock; |
| |
| import org.apache.hadoop.hbase.Cell; |
| 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.Increment; |
| import org.apache.hadoop.hbase.client.Result; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.phoenix.coprocessor.MetaDataProtocol; |
| import org.apache.phoenix.coprocessor.SequenceRegionObserver; |
| import org.apache.phoenix.exception.SQLExceptionCode; |
| import org.apache.phoenix.exception.SQLExceptionInfo; |
| import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; |
| import org.apache.phoenix.query.QueryConstants; |
| 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.KeyValueUtil; |
| import org.apache.phoenix.util.SequenceUtil; |
| |
| import com.google.common.collect.Lists; |
| import com.google.common.math.LongMath; |
| |
| public class Sequence { |
| public static final int SUCCESS = 0; |
| |
| public enum ValueOp {VALIDATE_SEQUENCE, INCREMENT_SEQUENCE}; |
| public enum MetaOp {CREATE_SEQUENCE, DROP_SEQUENCE, RETURN_SEQUENCE}; |
| |
| // create empty Sequence key values used while created a sequence row |
| private static final KeyValue CURRENT_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, CURRENT_VALUE_BYTES); |
| private static final KeyValue INCREMENT_BY_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, INCREMENT_BY_BYTES); |
| private static final KeyValue CACHE_SIZE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, CACHE_SIZE_BYTES); |
| private static final KeyValue MIN_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, MIN_VALUE_BYTES); |
| private static final KeyValue MAX_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, MAX_VALUE_BYTES); |
| private static final KeyValue CYCLE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, CYCLE_FLAG_BYTES); |
| private static final KeyValue LIMIT_REACHED_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SYSTEM_SEQUENCE_FAMILY_BYTES, LIMIT_REACHED_FLAG_BYTES); |
| private static final List<KeyValue> SEQUENCE_KV_COLUMNS = Arrays.<KeyValue>asList( |
| CURRENT_VALUE_KV, |
| INCREMENT_BY_KV, |
| CACHE_SIZE_KV, |
| // The following three columns were added in 3.1/4.1 |
| MIN_VALUE_KV, |
| MAX_VALUE_KV, |
| CYCLE_KV, |
| LIMIT_REACHED_KV |
| ); |
| static { |
| Collections.sort(SEQUENCE_KV_COLUMNS, KeyValue.COMPARATOR); |
| } |
| // Pre-compute index of sequence key values to prevent binary search |
| private static final int CURRENT_VALUE_INDEX = SEQUENCE_KV_COLUMNS.indexOf(CURRENT_VALUE_KV); |
| private static final int INCREMENT_BY_INDEX = SEQUENCE_KV_COLUMNS.indexOf(INCREMENT_BY_KV); |
| private static final int CACHE_SIZE_INDEX = SEQUENCE_KV_COLUMNS.indexOf(CACHE_SIZE_KV); |
| private static final int MIN_VALUE_INDEX = SEQUENCE_KV_COLUMNS.indexOf(MIN_VALUE_KV); |
| private static final int MAX_VALUE_INDEX = SEQUENCE_KV_COLUMNS.indexOf(MAX_VALUE_KV); |
| private static final int CYCLE_INDEX = SEQUENCE_KV_COLUMNS.indexOf(CYCLE_KV); |
| private static final int LIMIT_REACHED_INDEX = SEQUENCE_KV_COLUMNS.indexOf(LIMIT_REACHED_KV); |
| |
| public static final int NUM_SEQUENCE_KEY_VALUES = SEQUENCE_KV_COLUMNS.size(); |
| private static final EmptySequenceCacheException EMPTY_SEQUENCE_CACHE_EXCEPTION = new EmptySequenceCacheException(); |
| |
| private final SequenceKey key; |
| private final ReentrantLock lock; |
| private List<SequenceValue> values; |
| |
| public Sequence(SequenceKey key) { |
| if (key == null) throw new NullPointerException(); |
| this.key = key; |
| this.lock = new ReentrantLock(); |
| } |
| |
| private void insertSequenceValue(SequenceValue value) { |
| if (values == null) { |
| values = Lists.newArrayListWithExpectedSize(1); |
| values.add(value); |
| } else { |
| int i = values.size()-1; |
| while (i >= 0 && values.get(i).timestamp > value.timestamp) { |
| i--; |
| } |
| // Don't insert another value if there's one at the same timestamp that is a delete |
| if (i >= 0 && values.get(i).timestamp == value.timestamp) { |
| if (values.get(i).isDeleted) { |
| throw new IllegalStateException("Unexpected delete marker at timestamp " + value.timestamp + " for "+ key); |
| } |
| values.set(i, value); |
| } else { |
| values.add(i+1, value); |
| } |
| } |
| } |
| |
| private SequenceValue findSequenceValue(long timestamp) { |
| if (values == null) { |
| return null; |
| } |
| int i = values.size()-1; |
| while (i >= 0 && values.get(i).timestamp >= timestamp) { |
| i--; |
| } |
| if (i < 0) { |
| return null; |
| } |
| SequenceValue value = values.get(i); |
| return value.isDeleted ? null : value; |
| } |
| |
| private long increment(SequenceValue value, ValueOp op, long numToAllocate) throws SQLException { |
| boolean increasingSeq = value.incrementBy > 0 && op != ValueOp.VALIDATE_SEQUENCE; |
| // check if the the sequence has already reached the min/max limit |
| if (value.limitReached && op != ValueOp.VALIDATE_SEQUENCE) { |
| if (value.cycle) { |
| value.limitReached=false; |
| throw EMPTY_SEQUENCE_CACHE_EXCEPTION; |
| } else { |
| SQLExceptionCode code = |
| increasingSeq ? SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE |
| : SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE; |
| throw SequenceUtil.getException(this.key.getSchemaName(), |
| this.key.getSequenceName(), code); |
| } |
| } |
| |
| long returnValue = value.currentValue; |
| if (op == ValueOp.INCREMENT_SEQUENCE) { |
| boolean overflowOrUnderflow=false; |
| // advance currentValue while checking for overflow |
| try { |
| // advance by numToAllocate * the increment amount |
| value.currentValue = LongMath.checkedAdd(value.currentValue, numToAllocate * value.incrementBy); |
| } catch (ArithmeticException e) { |
| overflowOrUnderflow = true; |
| } |
| |
| // set the limitReached flag (which will be checked the next time increment is called) |
| // if overflow or limit was reached |
| if (overflowOrUnderflow || (increasingSeq && value.currentValue > value.maxValue) |
| || (!increasingSeq && value.currentValue < value.minValue)) { |
| value.limitReached=true; |
| } |
| } |
| return returnValue; |
| } |
| |
| public long incrementValue(long timestamp, ValueOp op, long numToAllocate) throws SQLException { |
| SequenceValue value = findSequenceValue(timestamp); |
| if (value == null) { |
| throw EMPTY_SEQUENCE_CACHE_EXCEPTION; |
| } |
| |
| if (isSequenceCacheExhausted(numToAllocate, value)) { |
| if (op == ValueOp.VALIDATE_SEQUENCE) { |
| return value.currentValue; |
| } |
| throw EMPTY_SEQUENCE_CACHE_EXCEPTION; |
| } |
| return increment(value, op, numToAllocate); |
| } |
| |
| /** |
| * This method first checks whether value.currentValue = value.nextValue, this check is what |
| * determines whether we need to refresh the cache when evaluating NEXT VALUE FOR. Once |
| * current value reaches the next value we know the cache is exhausted as we give sequence |
| * values out one at time. |
| * |
| * However for bulk allocations, evaluated by NEXT <n> VALUE FOR, we need a different check |
| * @see isSequenceCacheExhaustedForBulkAllocation |
| * |
| * Using the bulk allocation method for determining if the cache is exhausted for both cases |
| * works in most of the cases, however when dealing with CYCLEs and overflow and underflow, things |
| * break down due to things like sign changes that can happen if we overflow from a positive to |
| * a negative number and vice versa. Therefore, leaving both checks in place. |
| * |
| */ |
| private boolean isSequenceCacheExhausted(final long numToAllocate, final SequenceValue value) throws SQLException { |
| return value.currentValue == value.nextValue || (SequenceUtil.isBulkAllocation(numToAllocate) && isSequenceCacheExhaustedForBulkAllocation(numToAllocate, value)); |
| } |
| |
| /** |
| * This method checks whether there are sufficient values in the SequenceValue |
| * cached on the client to allocate the requested number of slots. It handles |
| * decreasing and increasing sequences as well as any overflows or underflows |
| * encountered. |
| */ |
| private boolean isSequenceCacheExhaustedForBulkAllocation(final long numToAllocate, final SequenceValue value) throws SQLException { |
| long targetSequenceValue; |
| |
| performValidationForBulkAllocation(numToAllocate, value); |
| |
| try { |
| targetSequenceValue = LongMath.checkedAdd(value.currentValue, numToAllocate * value.incrementBy); |
| } catch (ArithmeticException e) { |
| // Perform a CheckedAdd to make sure if over/underflow |
| // We don't treat this as the cache being exhausted as the current value may be valid in the case |
| // of no cycle, logic in increment() will take care of detecting we've hit the limit of the sequence |
| return false; |
| } |
| |
| if (value.incrementBy > 0) { |
| return targetSequenceValue > value.nextValue; |
| } else { |
| return targetSequenceValue < value.nextValue; |
| } |
| } |
| |
| /** |
| * @throws SQLException with the correct error code if sequence limit is reached with |
| * this request for allocation or we attempt to perform a bulk allocation on a sequence |
| * with cycles. |
| */ |
| private void performValidationForBulkAllocation(final long numToAllocate, final SequenceValue value) |
| throws SQLException { |
| boolean increasingSeq = value.incrementBy > 0 ? true : false; |
| |
| // We don't support Bulk Allocations on sequences that have the CYCLE flag set to true |
| // Check for this here so we fail on expression evaluation and don't allow corner case |
| // whereby a client requests less than cached number of slots on sequence with cycle to succeed |
| if (value.cycle && !SequenceUtil.isCycleAllowed(numToAllocate)) { |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.NUM_SEQ_TO_ALLOCATE_NOT_SUPPORTED) |
| .setSchemaName(key.getSchemaName()) |
| .setTableName(key.getSequenceName()) |
| .build().buildException(); |
| } |
| |
| if (SequenceUtil.checkIfLimitReached(value.currentValue, value.minValue, value.maxValue, value.incrementBy, value.cacheSize, numToAllocate)) { |
| throw new SQLExceptionInfo.Builder(SequenceUtil.getLimitReachedErrorCode(increasingSeq)) |
| .setSchemaName(key.getSchemaName()) |
| .setTableName(key.getSequenceName()) |
| .build().buildException(); |
| } |
| } |
| |
| public List<Append> newReturns() { |
| if (values == null) { |
| return Collections.emptyList(); |
| } |
| List<Append> appends = Lists.newArrayListWithExpectedSize(values.size()); |
| for (SequenceValue value : values) { |
| if (value.isInitialized() && value.currentValue != value.nextValue) { |
| appends.add(newReturn(value)); |
| } |
| } |
| return appends; |
| } |
| |
| public Append newReturn(long timestamp) throws EmptySequenceCacheException { |
| SequenceValue value = findSequenceValue(timestamp); |
| if (value == null) { |
| throw EMPTY_SEQUENCE_CACHE_EXCEPTION; |
| } |
| if (value.currentValue == value.nextValue) { |
| throw EMPTY_SEQUENCE_CACHE_EXCEPTION; |
| } |
| return newReturn(value); |
| } |
| |
| private Append newReturn(SequenceValue value) { |
| byte[] key = this.key.getKey(); |
| Append append = new Append(key); |
| byte[] opBuf = new byte[] {(byte)MetaOp.RETURN_SEQUENCE.ordinal()}; |
| append.setAttribute(SequenceRegionObserver.OPERATION_ATTRIB, opBuf); |
| append.setAttribute(SequenceRegionObserver.CURRENT_VALUE_ATTRIB, PLong.INSTANCE.toBytes(value.nextValue)); |
| Map<byte[], List<Cell>> familyMap = append.getFamilyCellMap(); |
| familyMap.put(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList( |
| (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, value.timestamp, PLong.INSTANCE.toBytes(value.currentValue)), |
| (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, value.timestamp, PBoolean.INSTANCE.toBytes(value.limitReached)) |
| )); |
| return append; |
| } |
| |
| public long currentValue(long timestamp) throws EmptySequenceCacheException { |
| SequenceValue value = findSequenceValue(timestamp); |
| if (value == null || value.isUnitialized()) { |
| throw EMPTY_SEQUENCE_CACHE_EXCEPTION; |
| } |
| return value.currentValue - value.incrementBy; |
| } |
| |
| public ReentrantLock getLock() { |
| return lock; |
| } |
| |
| public SequenceKey getKey() { |
| return key; |
| } |
| |
| public long incrementValue(Result result, ValueOp op, long numToAllocate) throws SQLException { |
| // In this case, we don't definitely know the timestamp of the deleted sequence, |
| // but we know anything older is likely deleted. Worse case, we remove a 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.rawCells().length == 1) { |
| Cell errorKV = result.rawCells()[0]; |
| int errorCode = PInteger.INSTANCE.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 |
| // like overkill. |
| // if (code == SQLExceptionCode.SEQUENCE_UNDEFINED) { |
| // } |
| throw new SQLExceptionInfo.Builder(code) |
| .setSchemaName(key.getSchemaName()) |
| .setTableName(key.getSequenceName()) |
| .build().buildException(); |
| } |
| // If we found the sequence, we update our cache with the new value |
| SequenceValue value = new SequenceValue(result, op, numToAllocate); |
| insertSequenceValue(value); |
| return increment(value, op, numToAllocate); |
| } |
| |
| |
| public Increment newIncrement(long timestamp, Sequence.ValueOp action, long numToAllocate) { |
| byte[] incKey = key.getKey(); |
| byte[] incValue = Bytes.toBytes((long)action.ordinal()); |
| Increment inc = new Increment(incKey); |
| // It doesn't matter what we set the amount too - we always use the values we get |
| // from the Get we do to prevent any race conditions. All columns that get added |
| // are returned with their current value |
| try { |
| inc.setTimeRange(MetaDataProtocol.MIN_TABLE_TIMESTAMP, timestamp); |
| inc.setAttribute(SequenceRegionObserver.NUM_TO_ALLOCATE, Bytes.toBytes(numToAllocate)); |
| } catch (IOException e) { |
| throw new RuntimeException(e); // Impossible |
| } |
| for (KeyValue kv : SEQUENCE_KV_COLUMNS) { |
| try { |
| // Store the timestamp on the cell as well as HBase 1.2 seems to not |
| // be serializing over the time range (see HBASE-15698). |
| Cell cell = new KeyValue(incKey, 0, incKey.length, |
| kv.getFamilyArray(), kv.getFamilyOffset(), kv.getFamilyLength(), |
| kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength(), |
| timestamp, |
| KeyValue.Type.Put, |
| incValue, 0, incValue.length); |
| inc.add(cell); |
| } catch (IOException e) { |
| throw new RuntimeException(e); // Impossible |
| } |
| } |
| return inc; |
| } |
| |
| /** |
| * Returns a KeyValue from the input result row |
| * @param kv an empty KeyValue used only to get the column family and column qualifier of the |
| * key value to be returned (if the sequence row is from a previous version) |
| * @param cellIndex index of the KeyValue to be returned (if the sequence row is from a previous version |
| * @return KeyValue |
| */ |
| private static KeyValue getKeyValue(Result r, KeyValue kv, int cellIndex) { |
| Cell[] cells = r.rawCells(); |
| // if the sequence row is from a previous version then MIN_VALUE, MAX_VALUE, CYCLE and LIMIT_REACHED key values are not present, |
| // the sequence row has only three columns (INCREMENT_BY, CACHE_SIZE and CURRENT_VALUE) and the order of the cells |
| // in the array returned by rawCells() is not what what we expect so use getColumnLatestCell() to get the cell we want |
| Cell cell = cells.length == NUM_SEQUENCE_KEY_VALUES ? cells[cellIndex] : |
| r.getColumnLatestCell(kv.getFamilyArray(), kv.getFamilyOffset(), kv.getFamilyLength(), kv.getQualifierArray(), kv.getQualifierOffset(), kv.getQualifierLength()); |
| return org.apache.hadoop.hbase.KeyValueUtil.ensureKeyValue(cell); |
| } |
| |
| private static KeyValue getKeyValue(Result r, KeyValue kv) { |
| return getKeyValue(r, kv, SEQUENCE_KV_COLUMNS.indexOf(kv)); |
| } |
| |
| public static KeyValue getCurrentValueKV(Result r) { |
| return getKeyValue(r, CURRENT_VALUE_KV, CURRENT_VALUE_INDEX); |
| } |
| |
| public static KeyValue getIncrementByKV(Result r) { |
| return getKeyValue(r, INCREMENT_BY_KV, INCREMENT_BY_INDEX); |
| } |
| |
| public static KeyValue getCacheSizeKV(Result r) { |
| return getKeyValue(r, CACHE_SIZE_KV, CACHE_SIZE_INDEX); |
| } |
| |
| public static KeyValue getMinValueKV(Result r) { |
| return getKeyValue(r, MIN_VALUE_KV, MIN_VALUE_INDEX); |
| } |
| |
| public static KeyValue getMaxValueKV(Result r) { |
| return getKeyValue(r, MAX_VALUE_KV, MAX_VALUE_INDEX); |
| } |
| |
| public static KeyValue getCycleKV(Result r) { |
| return getKeyValue(r, CYCLE_KV, CYCLE_INDEX); |
| } |
| |
| public static KeyValue getLimitReachedKV(Result r) { |
| return getKeyValue(r, LIMIT_REACHED_KV, LIMIT_REACHED_INDEX); |
| } |
| |
| public static void replaceCurrentValueKV(List<Cell> kvs, KeyValue currentValueKV) { |
| kvs.set(CURRENT_VALUE_INDEX, currentValueKV); |
| } |
| |
| public static void replaceMinValueKV(List<Cell> kvs, KeyValue minValueKV) { |
| kvs.set(MIN_VALUE_INDEX, minValueKV); |
| } |
| |
| public static void replaceMaxValueKV(List<Cell> kvs, KeyValue maxValueKV) { |
| kvs.set(MAX_VALUE_INDEX, maxValueKV); |
| } |
| |
| public static void replaceCycleValueKV(List<Cell> kvs, KeyValue cycleValueKV) { |
| kvs.set(CYCLE_INDEX, cycleValueKV); |
| } |
| public static void replaceLimitReachedKV(List<Cell> kvs, KeyValue limitReachedKV) { |
| kvs.set(LIMIT_REACHED_INDEX, limitReachedKV); |
| } |
| |
| /** |
| * Returns the KeyValues of r if it contains the expected number of KeyValues, |
| * else returns a list of KeyValues corresponding to SEQUENCE_KV_COLUMNS |
| */ |
| public static List<Cell> getCells(Result r, int numKVs) { |
| // if the sequence row is from a previous version |
| if (r.rawCells().length == numKVs ) |
| return Lists.newArrayList(r.rawCells()); |
| // else we need to handle missing MIN_VALUE, MAX_VALUE, CYCLE and LIMIT_REACHED KeyValues |
| List<Cell> cellList = Lists.newArrayListWithCapacity(NUM_SEQUENCE_KEY_VALUES); |
| for (KeyValue kv : SEQUENCE_KV_COLUMNS) { |
| cellList.add(getKeyValue(r,kv)); |
| } |
| return cellList; |
| } |
| |
| private static final class SequenceValue { |
| public final long incrementBy; |
| public final long timestamp; |
| public final long cacheSize; |
| |
| public long currentValue; |
| public long nextValue; |
| public long minValue; |
| public long maxValue; |
| public boolean cycle; |
| public boolean isDeleted; |
| public boolean limitReached; |
| |
| public SequenceValue(long timestamp, long minValue, long maxValue, boolean cycle) { |
| this(timestamp, false); |
| this.minValue = minValue; |
| this.maxValue = maxValue; |
| this.cycle = cycle; |
| } |
| |
| public SequenceValue(long timestamp, boolean isDeleted) { |
| this.timestamp = timestamp; |
| this.isDeleted = isDeleted; |
| this.incrementBy = 0; |
| this.limitReached = false; |
| this.cacheSize = 0; |
| } |
| |
| public boolean isInitialized() { |
| return this.incrementBy != 0; |
| } |
| |
| public boolean isUnitialized() { |
| return this.incrementBy == 0; |
| } |
| |
| public SequenceValue(Result r, ValueOp op, long numToAllocate) { |
| KeyValue currentValueKV = getCurrentValueKV(r); |
| KeyValue incrementByKV = getIncrementByKV(r); |
| KeyValue cacheSizeKV = getCacheSizeKV(r); |
| KeyValue minValueKV = getMinValueKV(r); |
| KeyValue maxValueKV = getMaxValueKV(r); |
| KeyValue cycleKV = getCycleKV(r); |
| this.timestamp = currentValueKV.getTimestamp(); |
| this.nextValue = PLong.INSTANCE.getCodec().decodeLong(currentValueKV.getValueArray(), currentValueKV.getValueOffset(), SortOrder.getDefault()); |
| this.incrementBy = PLong.INSTANCE.getCodec().decodeLong(incrementByKV.getValueArray(), incrementByKV.getValueOffset(), SortOrder.getDefault()); |
| this.cacheSize = PLong.INSTANCE.getCodec().decodeLong(cacheSizeKV.getValueArray(), cacheSizeKV.getValueOffset(), SortOrder.getDefault()); |
| this.minValue = PLong.INSTANCE.getCodec().decodeLong(minValueKV.getValueArray(), minValueKV.getValueOffset(), SortOrder.getDefault()); |
| this.maxValue = PLong.INSTANCE.getCodec().decodeLong(maxValueKV.getValueArray(), maxValueKV.getValueOffset(), SortOrder.getDefault()); |
| this.cycle = (Boolean) PBoolean.INSTANCE.toObject(cycleKV.getValueArray(), cycleKV.getValueOffset(), cycleKV.getValueLength()); |
| this.limitReached = false; |
| currentValue = nextValue; |
| |
| if (op != ValueOp.VALIDATE_SEQUENCE) { |
| // We can't just take the max of numToAllocate and cacheSize |
| // We need to handle a valid edgecase where a client requests bulk allocation of |
| // a number of slots that are less than cache size of the sequence |
| currentValue -= incrementBy * (SequenceUtil.isBulkAllocation(numToAllocate) ? numToAllocate : cacheSize); |
| } |
| } |
| } |
| |
| public boolean returnValue(Result result) throws SQLException { |
| Cell statusKV = result.rawCells()[0]; |
| if (statusKV.getValueLength() == 0) { // No error, but unable to return sequence values |
| return false; |
| } |
| long timestamp = statusKV.getTimestamp(); |
| int statusCode = PInteger.INSTANCE.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), SortOrder.getDefault()); |
| if (statusCode == SUCCESS) { // Success - update nextValue down to currentValue |
| SequenceValue value = findSequenceValue(timestamp); |
| if (value == null) { |
| throw new EmptySequenceCacheException(key.getSchemaName(),key.getSequenceName()); |
| } |
| return true; |
| } |
| SQLExceptionCode code = SQLExceptionCode.fromErrorCode(statusCode); |
| // TODO: We could have the server return the timestamps of the |
| // delete markers and we could insert them here, but this seems |
| // like overkill. |
| // if (code == SQLExceptionCode.SEQUENCE_UNDEFINED) { |
| // } |
| throw new SQLExceptionInfo.Builder(code) |
| .setSchemaName(key.getSchemaName()) |
| .setTableName(key.getSequenceName()) |
| .build().buildException(); |
| } |
| |
| public Append createSequence(long startWith, long incrementBy, long cacheSize, long timestamp, long minValue, long maxValue, boolean cycle) { |
| byte[] key = this.key.getKey(); |
| Append append = new Append(key); |
| append.setAttribute(SequenceRegionObserver.OPERATION_ATTRIB, new byte[] {(byte)MetaOp.CREATE_SEQUENCE.ordinal()}); |
| if (timestamp != HConstants.LATEST_TIMESTAMP) { |
| append.setAttribute(SequenceRegionObserver.MAX_TIMERANGE_ATTRIB, Bytes.toBytes(timestamp)); |
| } |
| Map<byte[], List<Cell>> familyMap = append.getFamilyCellMap(); |
| byte[] startWithBuf = PLong.INSTANCE.toBytes(startWith); |
| familyMap.put(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList( |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, timestamp, startWithBuf), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.START_WITH_BYTES, timestamp, startWithBuf), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.INCREMENT_BY_BYTES, timestamp, PLong.INSTANCE.toBytes(incrementBy)), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CACHE_SIZE_BYTES, timestamp, PLong.INSTANCE.toBytes(cacheSize)), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MIN_VALUE_BYTES, timestamp, PLong.INSTANCE.toBytes(minValue)), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MAX_VALUE_BYTES, timestamp, PLong.INSTANCE.toBytes(maxValue)), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CYCLE_FLAG_BYTES, timestamp, PBoolean.INSTANCE.toBytes(cycle)), |
| KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG_BYTES, timestamp, PDataType.FALSE_BYTES) |
| )); |
| return append; |
| } |
| |
| public long createSequence(Result result, long minValue, long maxValue, boolean cycle) throws SQLException { |
| Cell statusKV = result.rawCells()[0]; |
| long timestamp = statusKV.getTimestamp(); |
| int statusCode = PInteger.INSTANCE.getCodec().decodeInt(statusKV.getValueArray(), statusKV.getValueOffset(), SortOrder.getDefault()); |
| if (statusCode == 0) { // Success - add sequence value and return timestamp |
| SequenceValue value = new SequenceValue(timestamp, minValue, maxValue, cycle); |
| insertSequenceValue(value); |
| return timestamp; |
| } |
| SQLExceptionCode code = SQLExceptionCode.fromErrorCode(statusCode); |
| throw new SQLExceptionInfo.Builder(code) |
| .setSchemaName(key.getSchemaName()) |
| .setTableName(key.getSequenceName()) |
| .build().buildException(); |
| } |
| |
| public Append dropSequence(long timestamp) { |
| byte[] key = this.key.getKey(); |
| Append append = new Append(key); |
| append.setAttribute(SequenceRegionObserver.OPERATION_ATTRIB, new byte[] {(byte)MetaOp.DROP_SEQUENCE.ordinal()}); |
| if (timestamp != HConstants.LATEST_TIMESTAMP) { |
| append.setAttribute(SequenceRegionObserver.MAX_TIMERANGE_ATTRIB, Bytes.toBytes(timestamp)); |
| } |
| Map<byte[], List<Cell>> familyMap = append.getFamilyCellMap(); |
| familyMap.put(PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, Arrays.<Cell>asList( |
| (Cell)KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY))); |
| return append; |
| } |
| |
| public long dropSequence(Result result) throws SQLException { |
| Cell statusKV = result.rawCells()[0]; |
| long timestamp = statusKV.getTimestamp(); |
| int statusCode = PInteger.INSTANCE.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 |
| insertSequenceValue(new SequenceValue(timestamp, true)); |
| return timestamp; |
| } |
| // TODO: We could have the server return the timestamps of the |
| // delete markers and we could insert them here, but this seems |
| // like overkill. |
| // if (code == SQLExceptionCode.SEQUENCE_UNDEFINED) { |
| // } |
| throw new SQLExceptionInfo.Builder(code) |
| .setSchemaName(key.getSchemaName()) |
| .setTableName(key.getSequenceName()) |
| .build().buildException(); |
| } |
| |
| public static String getCreateTableStatement(int nSaltBuckets) { |
| if (nSaltBuckets <= 0) { |
| return QueryConstants.CREATE_SEQUENCE_METADATA; |
| } |
| return QueryConstants.CREATE_SEQUENCE_METADATA + "," + PhoenixDatabaseMetaData.SALT_BUCKETS + "=" + nSaltBuckets; |
| } |
| } |
