You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2014/07/20 01:41:09 UTC
[1/3] PHOENIX-1016 Support MINVALUE, MAXVALUE,
and CYCLE options in CREATE SEQUENCE (Thomas D'Silva)
Repository: phoenix
Updated Branches:
refs/heads/3.0 6d4c9f420 -> 866d43266
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index 7e81915..20ab137 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -28,6 +28,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_FAMILY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_SIZE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CURRENT_VALUE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CYCLE_FLAG;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DECIMAL_DIGITS;
@@ -41,6 +42,8 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_NULLABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_VIEW_REFERENCED;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.KEY_SEQ;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LINK_TYPE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MAX_VALUE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MIN_VALUE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MULTI_TENANT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NULLABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NUM_PREC_RADIX;
@@ -213,15 +216,19 @@ public interface QueryConstants {
HTableDescriptor.SPLIT_POLICY + "='" + MetaDataSplitPolicy.class.getName() + "'\n";
public static final String CREATE_SEQUENCE_METADATA =
- "CREATE TABLE IF NOT EXISTS " + SYSTEM_CATALOG_SCHEMA + ".\"" + TYPE_SEQUENCE + "\"(\n" +
+ "CREATE TABLE " + SYSTEM_CATALOG_SCHEMA + ".\"" + TYPE_SEQUENCE + "\"(\n" +
TENANT_ID + " VARCHAR NULL," +
- SEQUENCE_SCHEMA + " VARCHAR NULL, \n" +
+ SEQUENCE_SCHEMA + " VARCHAR NULL, \n" +
SEQUENCE_NAME + " VARCHAR NOT NULL, \n" +
START_WITH + " BIGINT, \n" +
- CURRENT_VALUE + " BIGINT, \n" +
+ CURRENT_VALUE + " BIGINT, \n" +
INCREMENT_BY + " BIGINT, \n" +
- CACHE_SIZE + " BIGINT \n" +
- " CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + "," + SEQUENCE_SCHEMA + "," + SEQUENCE_NAME + "))\n" +
- HConstants.VERSIONS + "=" + MetaDataProtocol.DEFAULT_MAX_META_DATA_VERSIONS + "\n";
+ CACHE_SIZE + " BIGINT, \n" +
+ // the following three columns were added in 3.0/4.0
+ MIN_VALUE + " BIGINT, \n" +
+ MAX_VALUE + " BIGINT, \n" +
+ CYCLE_FLAG + " BOOLEAN \n" +
+ " CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + "," + SEQUENCE_SCHEMA + "," + SEQUENCE_NAME + "))\n" +
+ HConstants.VERSIONS + "=" + MetaDataProtocol.DEFAULT_MAX_META_DATA_VERSIONS + "\n";
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index dc59d83..53aebf5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -637,7 +637,9 @@ public class MetaDataClient {
SequenceKey key = MetaDataUtil.getViewIndexSequenceKey(tenantIdStr, physicalName);
// Create at parent timestamp as we know that will be earlier than now
// and earlier than any SCN if one is set.
- createSequence(key.getTenantId(), key.getSchemaName(), key.getSequenceName(), true, Short.MIN_VALUE, 1, 1, dataTable.getTimeStamp());
+ createSequence(key.getTenantId(), key.getSchemaName(), key.getSequenceName(),
+ true, Short.MIN_VALUE, 1, 1, false, Long.MIN_VALUE, Long.MAX_VALUE,
+ dataTable.getTimeStamp());
long[] seqValues = new long[1];
SQLException[] sqlExceptions = new SQLException[1];
connection.getQueryServices().incrementSequences(Collections.singletonList(key), timestamp, seqValues, sqlExceptions);
@@ -697,16 +699,23 @@ public class MetaDataClient {
return new MutationState(1, connection);
}
- public MutationState createSequence(CreateSequenceStatement statement, long startWith, long incrementBy, long cacheSize) throws SQLException {
+ public MutationState createSequence(CreateSequenceStatement statement, long startWith,
+ long incrementBy, long cacheSize, long minValue, long maxValue) throws SQLException {
Long scn = connection.getSCN();
long timestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
- String tenantId = connection.getTenantId() == null ? null : connection.getTenantId().getString();
- return createSequence(tenantId, statement.getSequenceName().getSchemaName(), statement.getSequenceName().getTableName(), statement.ifNotExists(), startWith, incrementBy, cacheSize, timestamp);
+ String tenantId =
+ connection.getTenantId() == null ? null : connection.getTenantId().getString();
+ return createSequence(tenantId, statement.getSequenceName().getSchemaName(), statement
+ .getSequenceName().getTableName(), statement.ifNotExists(), startWith, incrementBy,
+ cacheSize, statement.getCycle(), minValue, maxValue, timestamp);
}
-
- private MutationState createSequence(String tenantId, String schemaName, String sequenceName, boolean ifNotExists, long startWith, long incrementBy, long cacheSize, long timestamp) throws SQLException {
+
+ private MutationState createSequence(String tenantId, String schemaName, String sequenceName,
+ boolean ifNotExists, long startWith, long incrementBy, long cacheSize, boolean cycle,
+ long minValue, long maxValue, long timestamp) throws SQLException {
try {
- connection.getQueryServices().createSequence(tenantId, schemaName, sequenceName, startWith, incrementBy, cacheSize, timestamp);
+ connection.getQueryServices().createSequence(tenantId, schemaName, sequenceName,
+ startWith, incrementBy, cacheSize, minValue, maxValue, cycle, timestamp);
} catch (SequenceAlreadyExistsException e) {
if (ifNotExists) {
return new MutationState(0, connection);
@@ -715,7 +724,7 @@ public class MetaDataClient {
}
return new MutationState(1, connection);
}
-
+
private static ColumnDef findColumnDefOrNull(List<ColumnDef> colDefs, ColumnName colName) {
for (ColumnDef colDef : colDefs) {
if (colDef.getColumnDefName().getColumnName().equals(colName.getColumnName())) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
index 19b5569..eff39ab 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/Sequence.java
@@ -20,8 +20,12 @@ 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.MAX_VALUE_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MIN_VALUE_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.START_WITH_BYTES;
import java.io.IOException;
import java.sql.SQLException;
@@ -46,8 +50,10 @@ import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.SchemaUtil;
+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;
@@ -55,23 +61,37 @@ public class Sequence {
public enum ValueOp {VALIDATE_SEQUENCE, RESERVE_SEQUENCE};
public enum MetaOp {CREATE_SEQUENCE, DROP_SEQUENCE, RETURN_SEQUENCE};
- // Pre-compute index of sequence key values to prevent binary search
+ // create empty Sequence key values used while created a sequence row
private static final KeyValue CURRENT_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, CURRENT_VALUE_BYTES);
private static final KeyValue INCREMENT_BY_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, INCREMENT_BY_BYTES);
private static final KeyValue CACHE_SIZE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, CACHE_SIZE_BYTES);
+ private static final KeyValue START_WITH_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, START_WITH_BYTES);
+ private static final KeyValue MIN_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, MIN_VALUE_BYTES);
+ private static final KeyValue MAX_VALUE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, MAX_VALUE_BYTES);
+ private static final KeyValue CYCLE_KV = KeyValue.createFirstOnRow(ByteUtil.EMPTY_BYTE_ARRAY, SEQUENCE_FAMILY_BYTES, CYCLE_FLAG_BYTES);
private static final List<KeyValue> SEQUENCE_KV_COLUMNS = Arrays.<KeyValue>asList(
CURRENT_VALUE_KV,
INCREMENT_BY_KV,
- CACHE_SIZE_KV
+ CACHE_SIZE_KV,
+ START_WITH_KV,
+ // the following three columns were added in 3.0/4.0
+ MIN_VALUE_KV,
+ MAX_VALUE_KV,
+ CYCLE_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 START_WITH_INDEX = SEQUENCE_KV_COLUMNS.indexOf(START_WITH_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 SEQUENCE_KEY_VALUES = SEQUENCE_KV_COLUMNS.size();
+ private 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;
@@ -95,7 +115,7 @@ public class Sequence {
}
// 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()) {
+ if (values.get(i).isDeleted) {
throw new IllegalStateException("Unexpected delete marker at timestamp " + value.timestamp + " for "+ key);
}
values.set(i, value);
@@ -117,23 +137,59 @@ public class Sequence {
return null;
}
SequenceValue value = values.get(i);
- return value.isDeleted() ? null : value;
+ return value.isDeleted ? null : value;
}
- public long incrementValue(long timestamp, int factor, ValueOp action) throws EmptySequenceCacheException {
+ private long increment(SequenceValue value, int factor) throws SQLException {
+ boolean increasingSeq = value.incrementBy > 0;
+ // check if the the sequence has already reached the min/max limit
+ if (value.limitReached) {
+ 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 (factor != 0) {
+ --value.unusedValues;
+ boolean overflowOrUnderflow=false;
+ // advance currentValue while checking for overflow
+ try {
+ long incrementValue = LongMath.checkedMultiply(value.incrementBy, factor);
+ value.currentValue = LongMath.checkedAdd(value.currentValue, incrementValue);
+ } 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, int factor, ValueOp action) throws SQLException {
SequenceValue value = findSequenceValue(timestamp);
if (value == null) {
throw EMPTY_SEQUENCE_CACHE_EXCEPTION;
}
- if (value.currentValue == value.nextValue) {
+ if (value.unusedValues == 0) {
if (action == ValueOp.VALIDATE_SEQUENCE) {
return value.currentValue;
}
throw EMPTY_SEQUENCE_CACHE_EXCEPTION;
- }
- long returnValue = value.currentValue;
- value.currentValue += factor * value.incrementBy;
- return returnValue;
+ }
+ return increment(value, factor);
}
public List<Append> newReturns() {
@@ -142,7 +198,7 @@ public class Sequence {
}
List<Append> appends = Lists.newArrayListWithExpectedSize(values.size());
for (SequenceValue value : values) {
- if (value.isInitialized() && value.currentValue != value.nextValue) {
+ if (value.isInitialized() && value.unusedValues>0) {
appends.add(newReturn(value));
}
}
@@ -154,7 +210,7 @@ public class Sequence {
if (value == null) {
throw EMPTY_SEQUENCE_CACHE_EXCEPTION;
}
- if (value.currentValue == value.nextValue) {
+ if (value.unusedValues==0) {
throw EMPTY_SEQUENCE_CACHE_EXCEPTION;
}
return newReturn(value);
@@ -165,10 +221,11 @@ public class Sequence {
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, PDataType.LONG.toBytes(value.nextValue));
+ append.setAttribute(SequenceRegionObserver.CURRENT_VALUE_ATTRIB, PDataType.LONG.toBytes(value.startValue));
Map<byte[], List<KeyValue>> familyMap = append.getFamilyMap();
familyMap.put(PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, Arrays.<KeyValue>asList(
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, value.timestamp, PDataType.LONG.toBytes(value.currentValue))
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, value.timestamp, ByteUtil.EMPTY_BYTE_ARRAY),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.START_WITH_BYTES, value.timestamp, PDataType.LONG.toBytes(value.currentValue))
));
return append;
}
@@ -212,9 +269,7 @@ public class Sequence {
// If we found the sequence, we update our cache with the new value
SequenceValue value = new SequenceValue(result);
insertSequenceValue(value);
- long currentValue = value.currentValue;
- value.currentValue += factor * value.incrementBy;
- return currentValue;
+ return increment(value, factor);
}
public Increment newIncrement(long timestamp, Sequence.ValueOp action) {
@@ -234,85 +289,140 @@ public class Sequence {
return inc;
}
- public static KeyValue getCurrentValueKV(List<KeyValue> kvs) {
- assert(kvs.size() == SEQUENCE_KEY_VALUES);
- return kvs.get(CURRENT_VALUE_INDEX);
- }
-
- public static KeyValue getIncrementByKV(List<KeyValue> kvs) {
- assert(kvs.size() == SEQUENCE_KEY_VALUES);
- return kvs.get(INCREMENT_BY_INDEX);
+ /**
+ * 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) {
+ KeyValue[] kvs = r.raw();
+ // if the sequence row is from a previous version then MIN_VALUE, MAX_VALUE and CYCLE key values are not present,
+ // the sequence row has only four columns (START_VALUE, 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
+ return kvs.length != NUM_SEQUENCE_KEY_VALUES ? r.getColumnLatest(kv.getFamily(), kv.getQualifier()) : (kvs[cellIndex]);
}
- public static KeyValue getCacheSizeKV(List<KeyValue> kvs) {
- assert(kvs.size() == SEQUENCE_KEY_VALUES);
- return kvs.get(CACHE_SIZE_INDEX);
+ private static KeyValue getKeyValue(Result r, KeyValue kv) {
+ return getKeyValue(r, kv, SEQUENCE_KV_COLUMNS.indexOf(kv));
}
public static KeyValue getCurrentValueKV(Result r) {
- KeyValue[] kvs = r.raw();
- assert(kvs.length == SEQUENCE_KEY_VALUES);
- return kvs[CURRENT_VALUE_INDEX];
+ return getKeyValue(r, CURRENT_VALUE_KV, CURRENT_VALUE_INDEX);
}
public static KeyValue getIncrementByKV(Result r) {
- KeyValue[] kvs = r.raw();
- assert(kvs.length == SEQUENCE_KEY_VALUES);
- return kvs[INCREMENT_BY_INDEX];
+ return getKeyValue(r, INCREMENT_BY_KV, INCREMENT_BY_INDEX);
}
public static KeyValue getCacheSizeKV(Result r) {
- KeyValue[] kvs = r.raw();
- assert(kvs.length == SEQUENCE_KEY_VALUES);
- return kvs[CACHE_SIZE_INDEX];
+ return getKeyValue(r, CACHE_SIZE_KV, CACHE_SIZE_INDEX);
}
- public static Result replaceCurrentValueKV(Result r, KeyValue currentValueKV) {
- KeyValue[] kvs = r.raw();
- List<KeyValue> newkvs = Lists.newArrayList(kvs);
- newkvs.set(CURRENT_VALUE_INDEX, currentValueKV);
- return new Result(newkvs);
+ public static KeyValue getStartValueKV(Result r) {
+ return getKeyValue(r, START_WITH_KV, START_WITH_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 void replaceCurrentValueKV(List<KeyValue> kvs, KeyValue currentValueKV) {
+ kvs.set(CURRENT_VALUE_INDEX, currentValueKV);
+ }
+
+ public static void replaceMinValueKV(List<KeyValue> kvs, KeyValue minValueKV) {
+ kvs.set(MIN_VALUE_INDEX, minValueKV);
+ }
+
+ public static void replaceMaxValueKV(List<KeyValue> kvs, KeyValue maxValueKV) {
+ kvs.set(MAX_VALUE_INDEX, maxValueKV);
+ }
+
+ public static void replaceCycleValueKV(List<KeyValue> kvs, KeyValue cycleValueKV) {
+ kvs.set(CYCLE_INDEX, cycleValueKV);
+ }
+
+ /**
+ * Returns a KeyValue[] for the result row. Handles empty MIN_VALUE, MAX_VALUE and CYCLE
+ * KeyValues if the sequence row is from a previous version
+ */
+ public static List<KeyValue> getCells(Result r) {
+ // if the sequence row is from a previous version
+ if (r.raw().length == NUM_SEQUENCE_KEY_VALUES )
+ return Lists.newArrayList(r.raw());
+ // else we need to handle missing MIN_VALUE, MAX_VALUE and CYCLE KeyValues
+ List<KeyValue> kvList = Lists.newArrayListWithCapacity(NUM_SEQUENCE_KEY_VALUES);
+ for (KeyValue kv : SEQUENCE_KV_COLUMNS) {
+ kvList.add(getKeyValue(r,kv));
+ }
+ return kvList;
}
private static final class SequenceValue {
public final long incrementBy;
- public final long cacheSize;
public final long timestamp;
public long currentValue;
- public long nextValue;
+ // start value of the current batch
+ public long startValue;
+ public long minValue;
+ public long maxValue;
+ public boolean cycle;
+ // number of values left in current batch
+ public long unusedValues;
+ public boolean isDeleted;
+ public boolean limitReached;
- public SequenceValue(long timestamp) {
+ 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.incrementBy = isDeleted ? -1 : 0;
- this.cacheSize = 0;
+ this.isDeleted = isDeleted;
+ this.incrementBy = 0;
+ this.limitReached = false;
}
public boolean isInitialized() {
- return this.incrementBy > 0;
+ return this.incrementBy != 0;
}
public boolean isUnitialized() {
return this.incrementBy == 0;
}
- public boolean isDeleted() {
- return this.incrementBy < 0;
- }
-
public SequenceValue(Result r) {
KeyValue currentValueKV = getCurrentValueKV(r);
KeyValue incrementByKV = getIncrementByKV(r);
KeyValue cacheSizeKV = getCacheSizeKV(r);
- timestamp = currentValueKV.getTimestamp();
- nextValue = PDataType.LONG.getCodec().decodeLong(currentValueKV.getBuffer(), currentValueKV.getValueOffset(), SortOrder.getDefault());
- incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getBuffer(), incrementByKV.getValueOffset(), SortOrder.getDefault());
- cacheSize = PDataType.LONG.getCodec().decodeLong(cacheSizeKV.getBuffer(), cacheSizeKV.getValueOffset(), SortOrder.getDefault());
- currentValue = nextValue - incrementBy * cacheSize;
+ KeyValue minValueKV = getMinValueKV(r);
+ KeyValue maxValueKV = getMaxValueKV(r);
+ KeyValue cycleKV = getCycleKV(r);
+ this.timestamp = currentValueKV.getTimestamp();
+ this.currentValue = PDataType.LONG.getCodec().decodeLong(currentValueKV.getBuffer(), currentValueKV.getValueOffset(), SortOrder.getDefault());
+ this.incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getBuffer(), incrementByKV.getValueOffset(), SortOrder.getDefault());
+ this.unusedValues = PDataType.LONG.getCodec().decodeLong(cacheSizeKV.getBuffer(), cacheSizeKV.getValueOffset(), SortOrder.getDefault());
+ this.minValue = PDataType.LONG.getCodec().decodeLong(minValueKV.getBuffer(), minValueKV.getValueOffset(), SortOrder.getDefault());
+ this.maxValue = PDataType.LONG.getCodec().decodeLong(maxValueKV.getBuffer(), maxValueKV.getValueOffset(), SortOrder.getDefault());
+ this.cycle = (Boolean)PDataType.BOOLEAN.toObject(cycleKV.getBuffer(), cycleKV.getValueOffset(), cycleKV.getValueLength());
+ this.limitReached = false;
+ // store the start value of this batch of sequence values, so that it can be used to
+ // determine if we can return unused sequence values when we close the connection
+ this.startValue = this.currentValue;
}
}
@@ -328,7 +438,6 @@ public class Sequence {
if (value == null) {
throw new EmptySequenceCacheException(key.getSchemaName(),key.getSequenceName());
}
- value.nextValue = value.currentValue;
return true;
}
SQLExceptionCode code = SQLExceptionCode.fromErrorCode(statusCode);
@@ -343,7 +452,7 @@ public class Sequence {
.build().buildException();
}
- public Append createSequence(long startWith, long incrementBy, long cacheSize, long timestamp) {
+ public Append createSequence(long startWith, long incrementBy, long cacheSize, long timestamp, long minValue, long maxValue, boolean cycle) {
byte[] key = SchemaUtil.getSequenceKey(this.key.getTenantId(), this.key.getSchemaName(), this.key.getSequenceName());
Append append = new Append(key);
append.setAttribute(SequenceRegionObserver.OPERATION_ATTRIB, new byte[] {(byte)MetaOp.CREATE_SEQUENCE.ordinal()});
@@ -354,20 +463,24 @@ public class Sequence {
byte[] startWithBuf = PDataType.LONG.toBytes(startWith);
familyMap.put(PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, Arrays.<KeyValue>asList(
KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, timestamp, startWithBuf),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CURRENT_VALUE_BYTES, timestamp, ByteUtil.EMPTY_BYTE_ARRAY),
KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.START_WITH_BYTES, timestamp, startWithBuf),
KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.INCREMENT_BY_BYTES, timestamp, PDataType.LONG.toBytes(incrementBy)),
- KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CACHE_SIZE_BYTES, timestamp, PDataType.LONG.toBytes(cacheSize))
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CACHE_SIZE_BYTES, timestamp, PDataType.LONG.toBytes(cacheSize)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MIN_VALUE_BYTES, timestamp, PDataType.LONG.toBytes(minValue)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.MAX_VALUE_BYTES, timestamp, PDataType.LONG.toBytes(maxValue)),
+ KeyValueUtil.newKeyValue(key, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES, PhoenixDatabaseMetaData.CYCLE_FLAG_BYTES, timestamp, PDataType.BOOLEAN.toBytes(cycle))
));
return append;
}
- public long createSequence(Result result) throws SQLException {
+
+ public long createSequence(Result result, long minValue, long maxValue, boolean cycle) throws SQLException {
KeyValue statusKV = result.raw()[0];
long timestamp = statusKV.getTimestamp();
int statusCode = PDataType.INTEGER.getCodec().decodeInt(statusKV.getBuffer(), statusKV.getValueOffset(), SortOrder.getDefault());
if (statusCode == 0) { // Success - add sequence value and return timestamp
- SequenceValue value = new SequenceValue(timestamp);
+ SequenceValue value = new SequenceValue(timestamp, minValue, maxValue, cycle);
insertSequenceValue(value);
return timestamp;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/schema/SequenceInfo.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/SequenceInfo.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/SequenceInfo.java
new file mode 100644
index 0000000..26ea132
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/SequenceInfo.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+public class SequenceInfo {
+ public long sequenceValue;
+ public final long incrementBy;
+ public final long minValue;
+ public final long maxValue;
+ public final long cacheSize;
+ public final boolean cycle;
+
+ public SequenceInfo(long sequenceValue, long incrementBy, long minValue, long maxValue, long cacheSize, boolean cycle) {
+ this.sequenceValue = sequenceValue;
+ this.incrementBy = incrementBy;
+ this.minValue = minValue;
+ this.maxValue = maxValue;
+ this.cacheSize = cacheSize;
+ this.cycle = cycle;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/util/KeyValueUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/KeyValueUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/KeyValueUtil.java
index d54aaea..2f10c15 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/KeyValueUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/KeyValueUtil.java
@@ -66,6 +66,24 @@ public class KeyValueUtil {
value, valueOffset, valueLength);
}
+ public static KeyValue newKeyValue(byte[] key, int keyOffset, int keyLength, byte[] cf,
+ int cfOffset, int cfLength, byte[] cq, int cqOffset, int cqLength, long ts, byte[] value,
+ int valueOffset, int valueLength) {
+ return new KeyValue(key, keyOffset, keyLength,
+ cf, cfOffset, cfLength,
+ cq, cqOffset, cqLength,
+ ts, Type.Put,
+ value, valueOffset, valueLength);
+ }
+
+ public static KeyValue newKeyValue(byte[] key, KeyValue kv, byte[] value) {
+ return newKeyValue(key, 0, key.length,
+ kv.getBuffer(), kv.getFamilyOffset(), kv.getFamilyLength(),
+ kv.getBuffer(), kv.getQualifierOffset(), kv.getQualifierLength(),
+ kv.getTimestamp(),
+ value, 0, value.length);
+ }
+
public static KeyValue newKeyValue(byte[] key, byte[] cf, byte[] cq, long ts, byte[] value) {
return newKeyValue(key,cf,cq,ts,value,0,value.length);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/util/SequenceUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/SequenceUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/SequenceUtil.java
new file mode 100644
index 0000000..f17d721
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SequenceUtil.java
@@ -0,0 +1,72 @@
+/*
+ * 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.util;
+
+import java.sql.SQLException;
+
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.schema.SequenceInfo;
+import org.apache.phoenix.schema.SequenceKey;
+
+import com.google.common.math.LongMath;
+
+/**
+ * Sequence related util functions
+ */
+public class SequenceUtil {
+
+ /**
+ * Returns the nextValue of a sequence
+ * @throws SQLException if cycle is false and the sequence limit has been reached
+ */
+ public static long getNextValue(long currentValue, long minValue, long maxValue,
+ long incrementBy, long cacheSize, boolean cycle) throws SQLException {
+ long nextValue = 0;
+ boolean increasingSeq = incrementBy > 0 ? true : false;
+ boolean overflowOrUnderflow = false;
+ // advance currentValue while checking for overflow
+ try {
+ long incrementValue = LongMath.checkedMultiply(incrementBy, cacheSize);
+ nextValue = LongMath.checkedAdd(currentValue, incrementValue);
+ } catch (ArithmeticException e) {
+ overflowOrUnderflow = true;
+ }
+
+ // check if overflow or limit was reached
+ if (overflowOrUnderflow || (increasingSeq && nextValue > maxValue)
+ || (!increasingSeq && nextValue < minValue)) {
+ if (cycle) {
+ nextValue = increasingSeq ? minValue : maxValue;
+ } else {
+ SQLExceptionCode code =
+ increasingSeq ? SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE
+ : SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE;
+ throw new SQLExceptionInfo.Builder(code).build().buildException();
+ }
+ }
+ return nextValue;
+ }
+
+ public static long getNextValue(SequenceKey key, SequenceInfo info) throws SQLException {
+ return getNextValue(info.sequenceValue, info.minValue, info.maxValue, info.incrementBy, info.cacheSize, info.cycle);
+ }
+
+ /**
+ * Helper function that returns a {@link SQLException}
+ */
+ public static SQLException getException(String schemaName, String tableName,
+ SQLExceptionCode code) {
+ return new SQLExceptionInfo.Builder(code).setSchemaName(schemaName).setTableName(tableName)
+ .build().buildException();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/test/java/org/apache/phoenix/compile/TenantSpecificViewIndexCompileTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/compile/TenantSpecificViewIndexCompileTest.java b/phoenix-core/src/test/java/org/apache/phoenix/compile/TenantSpecificViewIndexCompileTest.java
index e5cb2d6..94483b5 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/compile/TenantSpecificViewIndexCompileTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/compile/TenantSpecificViewIndexCompileTest.java
@@ -67,7 +67,7 @@ public class TenantSpecificViewIndexCompileTest extends BaseConnectionlessQueryT
conn.createStatement().execute("CREATE INDEX i1 ON v(v2)");
ResultSet rs = conn.createStatement().executeQuery("EXPLAIN SELECT v2 FROM v WHERE v2 > 'a' and k2 = 'a' ORDER BY v2,k2");
- assertEquals("CLIENT PARALLEL 1-WAY RANGE SCAN OVER _IDX_T ['me',-32768,'a'] - ['me',-32768,*]",
+ assertEquals("CLIENT PARALLEL 1-WAY RANGE SCAN OVER _IDX_T ['me',-32766,'a'] - ['me',-32766,*]",
QueryUtil.getExplainPlan(rs));
// Won't use index b/c v1 is not in index, but should optimize out k2 still from the order by
@@ -99,7 +99,7 @@ public class TenantSpecificViewIndexCompileTest extends BaseConnectionlessQueryT
// Confirm that a read-only view on an updatable view still optimizes out the read-olnly parts of the updatable view
ResultSet rs = conn.createStatement().executeQuery("EXPLAIN SELECT v2 FROM v2 WHERE v3 > 'a' and k2 = 'a' ORDER BY v3,k2");
- assertEquals("CLIENT PARALLEL 1-WAY RANGE SCAN OVER _IDX_T ['me',-32768,'a'] - ['me',-32768,*]",
+ assertEquals("CLIENT PARALLEL 1-WAY RANGE SCAN OVER _IDX_T ['me',-32767,'a'] - ['me',-32767,*]",
QueryUtil.getExplainPlan(rs));
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java
index 7d0755f..cfa8c4b 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java
@@ -86,5 +86,6 @@ public class PhoenixTestDriver extends PhoenixEmbeddedDriver {
connectionQueryServices.close();
queryServices.close();
}
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
index 51a834d..10c12ed 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
@@ -543,7 +543,8 @@ public abstract class BaseTest {
PhoenixTestDriver driver = new PhoenixTestDriver(props);
DriverManager.registerDriver(driver);
Assert.assertTrue(DriverManager.getDriver(url) == driver);
- driver.connect(url, TestUtil.TEST_PROPERTIES);
+ Connection conn = driver.connect(url, TestUtil.TEST_PROPERTIES);
+ conn.close();
return driver;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/test/java/org/apache/phoenix/util/SequenceUtilTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/SequenceUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/SequenceUtilTest.java
new file mode 100644
index 0000000..bb1ef49
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/util/SequenceUtilTest.java
@@ -0,0 +1,117 @@
+/**
+ * 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.util;
+
+import static org.junit.Assert.assertEquals;
+
+import java.sql.SQLException;
+
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.junit.Test;
+
+public class SequenceUtilTest {
+
+ private static long MIN_VALUE = 1;
+ private static long MAX_VALUE = 10;
+ private static long CACHE_SIZE = 2;
+
+ @Test
+ public void testAscendingNextValueWithinLimit() throws SQLException {
+ assertEquals(9, SequenceUtil.getNextValue(5, MIN_VALUE, MAX_VALUE, 2/* incrementBy */,
+ CACHE_SIZE, false));
+ }
+
+ @Test
+ public void testAscendingNextValueReachLimit() throws SQLException {
+ assertEquals(MAX_VALUE, SequenceUtil.getNextValue(6, MIN_VALUE, MAX_VALUE, 2/* incrementBy */,
+ CACHE_SIZE, false));
+ }
+
+ @Test
+ public void testAscendingNextValueGreaterThanMaxValueNoCycle() throws SQLException {
+ try {
+ SequenceUtil.getNextValue(MAX_VALUE, MIN_VALUE, MAX_VALUE, 2/* incrementBy */, CACHE_SIZE,
+ false);
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ }
+ }
+
+ @Test
+ public void testAscendingNextValueGreaterThanMaxValueCycle() throws SQLException {
+ assertEquals(MIN_VALUE, SequenceUtil.getNextValue(MAX_VALUE, MIN_VALUE, MAX_VALUE,
+ 2/* incrementBy */, CACHE_SIZE, true));
+ }
+
+ @Test
+ public void testAscendingOverflowNoCycle() throws SQLException {
+ try {
+ SequenceUtil.getNextValue(Long.MAX_VALUE, 0, Long.MAX_VALUE, 1/* incrementBy */, CACHE_SIZE,
+ false);
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ }
+ }
+
+ @Test
+ public void testAscendingOverflowCycle() throws SQLException {
+ assertEquals(0, SequenceUtil.getNextValue(Long.MAX_VALUE, 0, Long.MAX_VALUE,
+ 1/* incrementBy */, CACHE_SIZE, true));
+ }
+
+ @Test
+ public void testDescendingNextValueWithinLimit() throws SQLException {
+ assertEquals(2, SequenceUtil.getNextValue(6, MIN_VALUE, MAX_VALUE, -2/* incrementBy */,
+ CACHE_SIZE, false));
+ }
+
+ @Test
+ public void testDescendingNextValueReachLimit() throws SQLException {
+ assertEquals(MIN_VALUE, SequenceUtil.getNextValue(5, MIN_VALUE, MAX_VALUE, -2/* incrementBy */,
+ CACHE_SIZE, false));
+ }
+
+ @Test
+ public void testDescendingNextValueLessThanMinValueNoCycle() throws SQLException {
+ try {
+ SequenceUtil.getNextValue(1, MIN_VALUE, MAX_VALUE, -2/* incrementBy */, CACHE_SIZE,
+ false);
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE.getErrorCode(),
+ e.getErrorCode());
+ }
+ }
+
+ @Test
+ public void testDescendingNextValueLessThanMinValueCycle() throws SQLException {
+ assertEquals(MAX_VALUE, SequenceUtil.getNextValue(2, MIN_VALUE, MAX_VALUE,
+ -2/* incrementBy */, CACHE_SIZE, true));
+ }
+
+ @Test
+ public void testDescendingOverflowNoCycle() throws SQLException {
+ try {
+ SequenceUtil.getNextValue(Long.MIN_VALUE, Long.MIN_VALUE, 0, -1/* incrementBy */, CACHE_SIZE,
+ false);
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE.getErrorCode(),
+ e.getErrorCode());
+ }
+ }
+
+ @Test
+ public void testDescendingOverflowCycle() throws SQLException {
+ assertEquals(0, SequenceUtil.getNextValue(Long.MIN_VALUE, Long.MIN_VALUE, 0,
+ -1/* incrementBy */, CACHE_SIZE, true));
+ }
+}
[2/3] git commit: PHOENIX-1016 Support MINVALUE, MAXVALUE,
and CYCLE options in CREATE SEQUENCE (Thomas D'Silva)
Posted by ja...@apache.org.
PHOENIX-1016 Support MINVALUE, MAXVALUE, and CYCLE options in CREATE SEQUENCE (Thomas D'Silva)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/15e1108f
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/15e1108f
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/15e1108f
Branch: refs/heads/3.0
Commit: 15e1108fdde3abde0bba87ec2e40035f89868d30
Parents: 6d4c9f4
Author: James Taylor <jt...@salesforce.com>
Authored: Sat Jul 19 16:42:24 2014 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Sat Jul 19 16:42:24 2014 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/end2end/SequenceIT.java | 699 +++++++++++++++++--
phoenix-core/src/main/antlr3/PhoenixSQL.g | 13 +-
.../phoenix/compile/CreateSequenceCompiler.java | 161 +++--
.../coprocessor/SequenceRegionObserver.java | 114 ++-
.../phoenix/exception/SQLExceptionCode.java | 12 +-
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 7 +
.../apache/phoenix/jdbc/PhoenixStatement.java | 16 +-
.../phoenix/parse/CreateSequenceStatement.java | 45 +-
.../apache/phoenix/parse/ParseNodeFactory.java | 9 +-
.../apache/phoenix/parse/SelectStatement.java | 2 +-
.../phoenix/query/ConnectionQueryServices.java | 2 +-
.../query/ConnectionQueryServicesImpl.java | 31 +-
.../query/ConnectionlessQueryServicesImpl.java | 34 +-
.../query/DelegateConnectionQueryServices.java | 8 +-
.../apache/phoenix/query/QueryConstants.java | 19 +-
.../apache/phoenix/schema/MetaDataClient.java | 25 +-
.../org/apache/phoenix/schema/Sequence.java | 241 +++++--
.../org/apache/phoenix/schema/SequenceInfo.java | 29 +
.../org/apache/phoenix/util/KeyValueUtil.java | 18 +
.../org/apache/phoenix/util/SequenceUtil.java | 72 ++
.../TenantSpecificViewIndexCompileTest.java | 4 +-
.../apache/phoenix/jdbc/PhoenixTestDriver.java | 1 +
.../java/org/apache/phoenix/query/BaseTest.java | 3 +-
.../apache/phoenix/util/SequenceUtilTest.java | 117 ++++
24 files changed, 1428 insertions(+), 254 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
index 84bfece..1dbdb52 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SequenceIT.java
@@ -28,6 +28,7 @@ import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -39,16 +40,19 @@ import org.apache.phoenix.schema.SequenceNotFoundException;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.SequenceUtil;
import org.apache.phoenix.util.TestUtil;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@Category(ClientManagedTimeTest.class)
public class SequenceIT extends BaseClientManagedTimeIT {
+ private static final String NEXT_VAL_SQL = "SELECT NEXT VALUE FOR foo.bar FROM SYSTEM.\"SEQUENCE\"";
private static final long BATCH_SIZE = 3;
private Connection conn;
@@ -98,22 +102,22 @@ public class SequenceIT extends BaseClientManagedTimeIT {
}
}
-
+
@Test
- public void testCreateSequence() throws Exception {
+ public void testCreateSequence() throws Exception {
nextConnection();
- conn.createStatement().execute("CREATE SEQUENCE alpha.omega START WITH 2 INCREMENT BY 4");
+ conn.createStatement().execute("CREATE SEQUENCE alpha.omega START WITH 2 INCREMENT BY 4");
nextConnection();
- String query = "SELECT sequence_schema, sequence_name, current_value, increment_by FROM SYSTEM.\"SEQUENCE\" WHERE sequence_name='OMEGA'";
- ResultSet rs = conn.prepareStatement(query).executeQuery();
- assertTrue(rs.next());
- assertEquals("ALPHA", rs.getString("sequence_schema"));
- assertEquals("OMEGA", rs.getString("sequence_name"));
- assertEquals(2, rs.getInt("current_value"));
- assertEquals(4, rs.getInt("increment_by"));
- assertFalse(rs.next());
+ String query = "SELECT sequence_schema, sequence_name, current_value, increment_by FROM SYSTEM.\"SEQUENCE\" WHERE sequence_name='OMEGA'";
+ ResultSet rs = conn.prepareStatement(query).executeQuery();
+ assertTrue(rs.next());
+ assertEquals("ALPHA", rs.getString("sequence_schema"));
+ assertEquals("OMEGA", rs.getString("sequence_name"));
+ assertEquals(null, rs.getBytes("current_value"));
+ assertEquals(4, rs.getInt("increment_by"));
+ assertFalse(rs.next());
}
-
+
@Test
public void testCurrentValueFor() throws Exception {
ResultSet rs;
@@ -126,6 +130,7 @@ public class SequenceIT extends BaseClientManagedTimeIT {
fail();
} catch (SQLException e) {
assertEquals(SQLExceptionCode.CANNOT_CALL_CURRENT_BEFORE_NEXT_VALUE.getErrorCode(), e.getErrorCode());
+ assertTrue(e.getNextException()==null);
}
rs = conn.createStatement().executeQuery("SELECT NEXT VALUE FOR used.nowhere FROM SYSTEM.\"SEQUENCE\"");
@@ -146,7 +151,7 @@ public class SequenceIT extends BaseClientManagedTimeIT {
assertTrue(rs.next());
assertEquals("ALPHA", rs.getString("sequence_schema"));
assertEquals("OMEGA", rs.getString("sequence_name"));
- assertEquals(2, rs.getInt("current_value"));
+ assertEquals(null, rs.getBytes("current_value"));
assertEquals(4, rs.getInt("increment_by"));
assertFalse(rs.next());
@@ -168,18 +173,7 @@ public class SequenceIT extends BaseClientManagedTimeIT {
nextConnection();
conn.createStatement().execute("CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY 2");
nextConnection();
- String query = "SELECT NEXT VALUE FOR foo.bar FROM SYSTEM.\"SEQUENCE\"";
- ResultSet rs = conn.prepareStatement(query).executeQuery();
- assertTrue(rs.next());
- assertEquals(3, rs.getInt(1));
-
- rs = conn.prepareStatement(query).executeQuery();
- assertTrue(rs.next());
- assertEquals(5, rs.getInt(1));
-
- rs = conn.prepareStatement(query).executeQuery();
- assertTrue(rs.next());
- assertEquals(7, rs.getInt(1));
+ assertSequenceValuesForSingleRow(3, 5, 7);
}
@Test
@@ -200,30 +194,45 @@ public class SequenceIT extends BaseClientManagedTimeIT {
assertEquals(3, rs.getInt(1));
}
- @Test
- public void testSequenceCreation() throws Exception {
+ @Test
+ public void testSequenceCreation() throws Exception {
nextConnection();
- conn.createStatement().execute("CREATE SEQUENCE alpha.gamma START WITH 2 INCREMENT BY 3 CACHE 5");
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE alpha.gamma START WITH 2 INCREMENT BY 3 MINVALUE 0 MAXVALUE 10 CYCLE CACHE 5");
nextConnection();
- ResultSet rs = conn.createStatement().executeQuery("SELECT start_with, increment_by, cache_size, sequence_schema, sequence_name FROM SYSTEM.\"SEQUENCE\"");
+ ResultSet rs =
+ conn.createStatement()
+ .executeQuery(
+ "SELECT start_with, current_value, increment_by, cache_size, min_value, max_value, cycle_flag, sequence_schema, sequence_name FROM SYSTEM.\"SEQUENCE\"");
assertTrue(rs.next());
- assertEquals(2, rs.getLong(1));
- assertEquals(3, rs.getLong(2));
- assertEquals(5, rs.getLong(3));
- assertEquals("ALPHA", rs.getString(4));
- assertEquals("GAMMA", rs.getString(5));
+ assertEquals(2, rs.getLong("start_with"));
+ assertEquals(null, rs.getBytes("current_value"));
+ assertEquals(3, rs.getLong("increment_by"));
+ assertEquals(5, rs.getLong("cache_size"));
+ assertEquals(0, rs.getLong("min_value"));
+ assertEquals(10, rs.getLong("max_value"));
+ assertEquals(true, rs.getBoolean("cycle_flag"));
+ assertEquals("ALPHA", rs.getString("sequence_schema"));
+ assertEquals("GAMMA", rs.getString("sequence_name"));
assertFalse(rs.next());
- rs = conn.createStatement().executeQuery("SELECT NEXT VALUE FOR alpha.gamma, CURRENT VALUE FOR alpha.gamma FROM SYSTEM.\"SEQUENCE\"");
+ rs =
+ conn.createStatement()
+ .executeQuery(
+ "SELECT NEXT VALUE FOR alpha.gamma, CURRENT VALUE FOR alpha.gamma FROM SYSTEM.\"SEQUENCE\"");
assertTrue(rs.next());
assertEquals(2, rs.getLong(1));
assertEquals(2, rs.getLong(2));
assertFalse(rs.next());
- rs = conn.createStatement().executeQuery("SELECT CURRENT VALUE FOR alpha.gamma, NEXT VALUE FOR alpha.gamma FROM SYSTEM.\"SEQUENCE\"");
+ rs =
+ conn.createStatement()
+ .executeQuery(
+ "SELECT CURRENT VALUE FOR alpha.gamma, NEXT VALUE FOR alpha.gamma FROM SYSTEM.\"SEQUENCE\"");
assertTrue(rs.next());
assertEquals(5, rs.getLong(1));
assertEquals(5, rs.getLong(2));
assertFalse(rs.next());
- }
+ }
@Test
public void testSameMultipleSequenceValues() throws Exception {
@@ -240,16 +249,16 @@ public class SequenceIT extends BaseClientManagedTimeIT {
}
@Test
- public void testMultipleSequenceValues() throws Exception {
+ public void testMultipleSequenceValues() throws Exception {
nextConnection();
- conn.createStatement().execute("CREATE SEQUENCE alpha.zeta START WITH 4 INCREMENT BY 7");
- conn.createStatement().execute("CREATE SEQUENCE alpha.kappa START WITH 9 INCREMENT BY 2");
+ conn.createStatement().execute("CREATE SEQUENCE alpha.zeta START WITH 4 INCREMENT BY 7");
+ conn.createStatement().execute("CREATE SEQUENCE alpha.kappa START WITH 9 INCREMENT BY 2");
nextConnection();
- String query = "SELECT NEXT VALUE FOR alpha.zeta, NEXT VALUE FOR alpha.kappa FROM SYSTEM.\"SEQUENCE\"";
- ResultSet rs = conn.prepareStatement(query).executeQuery();
- assertTrue(rs.next());
- assertEquals(4, rs.getInt(1));
- assertEquals(9, rs.getInt(2));
+ String query = "SELECT NEXT VALUE FOR alpha.zeta, NEXT VALUE FOR alpha.kappa FROM SYSTEM.\"SEQUENCE\"";
+ ResultSet rs = conn.prepareStatement(query).executeQuery();
+ assertTrue(rs.next());
+ assertEquals(4, rs.getInt(1));
+ assertEquals(9, rs.getInt(2));
assertTrue(rs.next());
assertEquals(4+7, rs.getInt(1));
assertEquals(9+2, rs.getInt(2));
@@ -266,8 +275,79 @@ public class SequenceIT extends BaseClientManagedTimeIT {
assertEquals(9+2*3, rs.getInt(2));
assertFalse(rs.next());
conn.close();
- }
-
+ }
+
+ @Test
+ public void testMultipleSequencesNoCycle() throws Exception {
+ nextConnection();
+ conn.createStatement().execute(
+ "CREATE SEQUENCE alpha.zeta START WITH 4 INCREMENT BY 7 MAXVALUE 24");
+ conn.createStatement().execute(
+ "CREATE SEQUENCE alpha.kappa START WITH 9 INCREMENT BY -2 MINVALUE 5");
+ nextConnection();
+ String query =
+ "SELECT NEXT VALUE FOR alpha.zeta, NEXT VALUE FOR alpha.kappa FROM SYSTEM.\"SEQUENCE\"";
+ ResultSet rs = conn.prepareStatement(query).executeQuery();
+ assertTrue(rs.next());
+ assertEquals(4, rs.getInt(1));
+ assertEquals(9, rs.getInt(2));
+ assertTrue(rs.next());
+ assertEquals(4 + 7, rs.getInt(1));
+ assertEquals(9 - 2, rs.getInt(2));
+ assertFalse(rs.next());
+ conn.close();
+
+ nextConnection();
+ rs = conn.prepareStatement(query).executeQuery();
+ assertTrue(rs.next());
+ assertEquals(4 + 7 * 2, rs.getInt(1));
+ assertEquals(9 - 2 * 2, rs.getInt(2));
+ try {
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ SQLException sqlEx1 =
+ SequenceUtil.getException("ALPHA", "ZETA",
+ SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE);
+ SQLException sqlEx2 =
+ SequenceUtil.getException("ALPHA", "KAPPA",
+ SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE);
+ verifyExceptions(e, Lists.newArrayList(sqlEx1.getMessage(), sqlEx2.getMessage()));
+ }
+ conn.close();
+ }
+
+ @Test
+ public void testMultipleSequencesCycle() throws Exception {
+ nextConnection();
+ conn.createStatement().execute(
+ "CREATE SEQUENCE alpha.zeta START WITH 4 INCREMENT BY 7 MINVALUE 4 MAXVALUE 19 CYCLE");
+ conn.createStatement().execute(
+ "CREATE SEQUENCE alpha.kappa START WITH 9 INCREMENT BY -2 MINVALUE 5 MAXVALUE 9 CYCLE");
+ nextConnection();
+ String query =
+ "SELECT NEXT VALUE FOR alpha.zeta, NEXT VALUE FOR alpha.kappa FROM SYSTEM.\"SEQUENCE\"";
+ ResultSet rs = conn.prepareStatement(query).executeQuery();
+ assertTrue(rs.next());
+ assertEquals(4, rs.getInt(1));
+ assertEquals(9, rs.getInt(2));
+ assertTrue(rs.next());
+ assertEquals(4 + 7, rs.getInt(1));
+ assertEquals(9 - 2, rs.getInt(2));
+ assertFalse(rs.next());
+ conn.close();
+
+ nextConnection();
+ rs = conn.prepareStatement(query).executeQuery();
+ assertTrue(rs.next());
+ assertEquals(4 + 7 * 2, rs.getInt(1));
+ assertEquals(9 - 2 * 2, rs.getInt(2));
+ assertTrue(rs.next());
+ assertEquals(4, rs.getInt(1));
+ assertEquals(9, rs.getInt(2));
+ conn.close();
+ }
+
@Test
public void testCompilerOptimization() throws Exception {
nextConnection();
@@ -459,8 +539,8 @@ public class SequenceIT extends BaseClientManagedTimeIT {
stmt1.execute();
}
conn1.commit();
- conn1.close(); // will return unused sequences, so no gaps now
-
+ conn1.close();
+
nextConnection();
Connection conn2 = conn;
conn = null; // So that call to nextConnection doesn't close it
@@ -470,7 +550,7 @@ public class SequenceIT extends BaseClientManagedTimeIT {
stmt2.execute();
}
conn2.commit();
- conn1.close();
+ conn2.close();
nextConnection();
ResultSet rs = conn.createStatement().executeQuery("SELECT k FROM foo");
@@ -573,7 +653,7 @@ public class SequenceIT extends BaseClientManagedTimeIT {
rs = conn.createStatement().executeQuery("SELECT sequence_name, current_value FROM SYSTEM.\"SEQUENCE\" WHERE sequence_name='BAR'");
assertTrue(rs.next());
assertEquals("BAR", rs.getString(1));
- assertEquals(1, rs.getLong(2));
+ assertEquals(null, rs.getBytes(2));
conn.close();
conn2.close();
@@ -592,10 +672,10 @@ public class SequenceIT extends BaseClientManagedTimeIT {
nextConnection();
conn.createStatement().execute("CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY 2");
nextConnection();
- String query = "SELECT COALESCE(NEXT VALUE FOR foo.bar,1) FROM SYSTEM.\"SEQUENCE\"";
+ String query = "SELECT LPAD(ENCODE(NEXT VALUE FOR foo.bar,'base62'),5,'0') FROM SYSTEM.\"SEQUENCE\"";
ResultSet rs = conn.prepareStatement(query).executeQuery();
assertTrue(rs.next());
- assertEquals(3, rs.getInt(1));
+ assertEquals("00003", rs.getString(1));
}
@Test
@@ -609,11 +689,510 @@ public class SequenceIT extends BaseClientManagedTimeIT {
assertEquals(4, rs.getInt(1));
}
- private void nextConnection() throws Exception {
- if (conn != null) conn.close();
- long ts = nextTimestamp();
- Properties props = new Properties(TestUtil.TEST_PROPERTIES);
- props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts));
- conn = DriverManager.getConnection(getUrl(), props);
- }
+ private void nextConnection() throws Exception {
+ if (conn != null) conn.close();
+ long ts = nextTimestamp();
+ Properties props = new Properties(TestUtil.TEST_PROPERTIES);
+ props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts));
+ conn = DriverManager.getConnection(getUrl(), props);
+ }
+
+ @Test
+ public void testSequenceDefault() throws Exception {
+ nextConnection();
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar");
+ nextConnection();
+ assertSequenceValuesForSingleRow(1, 2, 3);
+ conn.createStatement().execute("DROP SEQUENCE foo.bar");
+
+ nextConnection();
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar INCREMENT BY -1");
+ nextConnection();
+ assertSequenceValuesForSingleRow(1, 0, -1);
+ conn.createStatement().execute("DROP SEQUENCE foo.bar");
+
+ nextConnection();
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar MINVALUE 10");
+ nextConnection();
+ assertSequenceValuesForSingleRow(10, 11, 12);
+ conn.createStatement().execute("DROP SEQUENCE foo.bar");
+
+ nextConnection();
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar INCREMENT BY -1 MINVALUE 10 ");
+ nextConnection();
+ assertSequenceValuesForSingleRow(Long.MAX_VALUE, Long.MAX_VALUE - 1, Long.MAX_VALUE - 2);
+ conn.createStatement().execute("DROP SEQUENCE foo.bar");
+
+ nextConnection();
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar MAXVALUE 0");
+ nextConnection();
+ assertSequenceValuesForSingleRow(Long.MIN_VALUE, Long.MIN_VALUE + 1, Long.MIN_VALUE + 2);
+ conn.createStatement().execute("DROP SEQUENCE foo.bar");
+
+ nextConnection();
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar INCREMENT BY -1 MAXVALUE 0");
+ nextConnection();
+ assertSequenceValuesForSingleRow(0, -1, -2);
+ }
+
+ @Test
+ public void testSequenceValidateStartValue() throws Exception {
+ nextConnection();
+ try {
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar1 START WITH 1 INCREMENT BY 1 MINVALUE 2 MAXVALUE 3");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.STARTS_WITH_MUST_BE_BETWEEN_MIN_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+
+ try {
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar2 START WITH 4 INCREMENT BY 1 MINVALUE 2 MAXVALUE 3");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.STARTS_WITH_MUST_BE_BETWEEN_MIN_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceValidateMinValue() throws Exception {
+ nextConnection();
+ try {
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar MINVALUE abc");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.MINVALUE_MUST_BE_CONSTANT.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceValidateMaxValue() throws Exception {
+ nextConnection();
+ try {
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar MAXVALUE null");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.MAXVALUE_MUST_BE_CONSTANT.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceValidateMinValueLessThanOrEqualToMaxValue() throws Exception {
+ nextConnection();
+ try {
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar MINVALUE 2 MAXVALUE 1");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(
+ SQLExceptionCode.MINVALUE_MUST_BE_LESS_THAN_OR_EQUAL_TO_MAXVALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceValidateIncrementConstant() throws Exception {
+ nextConnection();
+ try {
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar INCREMENT null");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.INCREMENT_BY_MUST_BE_CONSTANT.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceValidateIncrementNotEqualToZero() throws Exception {
+ nextConnection();
+ try {
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar INCREMENT 0");
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.INCREMENT_BY_MUST_NOT_BE_ZERO.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceStartWithMinMaxSameValueIncreasingCycle() throws Exception {
+ nextConnection();
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY 1 MINVALUE 3 MAXVALUE 3 CYCLE CACHE 1");
+ nextConnection();
+ assertSequenceValuesForSingleRow(3, 3, 3);
+ }
+
+ @Test
+ public void testSequenceStartWithMinMaxSameValueDecreasingCycle() throws Exception {
+ nextConnection();
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY -1 MINVALUE 3 MAXVALUE 3 CYCLE CACHE 2");
+ nextConnection();
+ assertSequenceValuesForSingleRow(3, 3, 3);
+ }
+
+ @Test
+ public void testSequenceStartWithMinMaxSameValueIncreasingNoCycle() throws Exception {
+ nextConnection();
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY 1 MINVALUE 3 MAXVALUE 3 CACHE 1");
+ nextConnection();
+ assertSequenceValuesForSingleRow(3);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceStartWithMinMaxSameValueDecreasingNoCycle() throws Exception {
+ nextConnection();
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY -1 MINVALUE 3 MAXVALUE 3 CACHE 2");
+ nextConnection();
+ assertSequenceValuesForSingleRow(3);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceIncreasingCycle() throws Exception {
+ nextConnection();
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 2 INCREMENT BY 3 MINVALUE 1 MAXVALUE 10 CYCLE CACHE 2");
+ nextConnection();
+ assertSequenceValuesForSingleRow(2, 5, 8, 1, 4, 7, 10, 1, 4);
+ }
+
+ @Test
+ public void testSequenceDecreasingCycle() throws Exception {
+ nextConnection();
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY -2 MINVALUE 1 MAXVALUE 10 CYCLE CACHE 2");
+ nextConnection();
+ assertSequenceValuesForSingleRow(3, 1, 10, 8, 6, 4, 2, 10, 8);
+ }
+
+ @Test
+ public void testSequenceIncreasingNoCycle() throws Exception {
+ nextConnection();
+ // client throws exception
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar START WITH 2 INCREMENT BY 3 MINVALUE 1 MAXVALUE 10 CACHE 100");
+ nextConnection();
+ assertSequenceValuesForSingleRow(2, 5, 8);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceIncreasingUsingMaxValueNoCycle() throws Exception {
+ nextConnection();
+ // server throws exception
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar START WITH 8 INCREMENT BY 2 MINVALUE 1 MAXVALUE 10 CACHE 2");
+ nextConnection();
+ assertSequenceValuesForSingleRow(8, 10);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceDecreasingNoCycle() throws Exception {
+ nextConnection();
+ // client will throw exception
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 4 INCREMENT BY -2 MINVALUE 1 MAXVALUE 10 CACHE 100");
+ nextConnection();
+ assertSequenceValuesForSingleRow(4, 2);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceDecreasingUsingMinValueNoCycle() throws Exception {
+ nextConnection();
+ // server will throw exception
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar START WITH 3 INCREMENT BY -2 MINVALUE 1 MAXVALUE 10 CACHE 2");
+ nextConnection();
+ assertSequenceValuesForSingleRow(3, 1);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceIncreasingOverflowNoCycle() throws Exception {
+ nextConnection();
+ // start with Long.MAX_VALUE
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar START WITH 9223372036854775807 INCREMENT BY 1 CACHE 10");
+ nextConnection();
+ assertSequenceValuesForSingleRow(Long.MAX_VALUE);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceIncreasingOverflowCycle() throws Exception {
+ nextConnection();
+ // start with Long.MAX_VALUE
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 9223372036854775807 INCREMENT BY 9223372036854775807 CYCLE CACHE 10");
+ nextConnection();
+ assertSequenceValuesForSingleRow(Long.MAX_VALUE, Long.MIN_VALUE, -1, Long.MAX_VALUE - 1,
+ Long.MIN_VALUE, -1);
+ }
+
+ @Test
+ public void testSequenceDecreasingOverflowNoCycle() throws Exception {
+ nextConnection();
+ // start with Long.MIN_VALUE + 1
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar START WITH -9223372036854775807 INCREMENT BY -1 CACHE 10");
+ nextConnection();
+ assertSequenceValuesForSingleRow(Long.MIN_VALUE + 1, Long.MIN_VALUE);
+ try {
+ ResultSet rs = conn.createStatement().executeQuery(NEXT_VAL_SQL);
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MIN_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testSequenceDecreasingOverflowCycle() throws Exception {
+ nextConnection();
+ // start with Long.MIN_VALUE + 1
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH -9223372036854775807 INCREMENT BY -9223372036854775807 CYCLE CACHE 10");
+ nextConnection();
+ assertSequenceValuesForSingleRow(Long.MIN_VALUE + 1, Long.MAX_VALUE, 0, Long.MIN_VALUE + 1,
+ Long.MAX_VALUE, 0);
+ }
+
+ @Test
+ public void testMultipleSequenceValuesNoCycle() throws Exception {
+ nextConnection();
+ conn.createStatement().execute(
+ "CREATE SEQUENCE foo.bar START WITH 1 INCREMENT BY 2 MINVALUE 1 MAXVALUE 10 CACHE 2");
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar2");
+ nextConnection();
+ assertSequenceValuesMultipleSeq(1, 3);
+ assertSequenceValuesMultipleSeq(5, 7);
+
+ ResultSet rs = conn.prepareStatement(NEXT_VAL_SQL).executeQuery();
+ assertTrue(rs.next());
+ assertEquals(9, rs.getInt(1));
+ try {
+ assertTrue(rs.next());
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+
+ try {
+ rs = conn.prepareStatement(NEXT_VAL_SQL).executeQuery();
+ rs.next();
+ fail();
+ } catch (SQLException e) {
+ assertEquals(SQLExceptionCode.SEQUENCE_VAL_REACHED_MAX_VALUE.getErrorCode(),
+ e.getErrorCode());
+ assertTrue(e.getNextException() == null);
+ }
+ }
+
+ @Test
+ public void testMultipleSequenceValuesCycle() throws Exception {
+ nextConnection();
+ conn.createStatement()
+ .execute(
+ "CREATE SEQUENCE foo.bar START WITH 1 INCREMENT BY 2 MINVALUE 1 MAXVALUE 10 CYCLE CACHE 2");
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar2");
+ nextConnection();
+ assertSequenceValuesMultipleSeq(1, 3);
+ assertSequenceValuesMultipleSeq(5, 7);
+ assertSequenceValuesMultipleSeq(9, 1);
+ assertSequenceValuesMultipleSeq(3, 5);
+ assertSequenceValuesMultipleSeq(7, 9);
+ assertSequenceValuesMultipleSeq(1, 3);
+ assertSequenceValuesMultipleSeq(5, 7);
+ }
+
+ @Test
+ public void testUpsertSelectGroupByWithSequence() throws Exception {
+ nextConnection();
+ conn.createStatement().execute("CREATE SEQUENCE foo.bar");
+ nextConnection();
+
+ conn.createStatement()
+ .execute(
+ "CREATE TABLE EVENTS (event_id BIGINT NOT NULL PRIMARY KEY, user_id char(15), val BIGINT )");
+ conn.createStatement()
+ .execute(
+ "CREATE TABLE METRICS (metric_id char(15) NOT NULL PRIMARY KEY, agg_id char(15), metric_val INTEGER )");
+
+ nextConnection();
+ // 2 rows for user1, 3 rows for user2 and 1 row for user3
+ insertEvent(1, "user1", 1);
+ insertEvent(2, "user2", 1);
+ insertEvent(3, "user1", 1);
+ insertEvent(4, "user2", 1);
+ insertEvent(5, "user2", 1);
+ insertEvent(6, "user3", 1);
+ conn.commit();
+ nextConnection();
+
+ conn.createStatement()
+ .execute(
+ "UPSERT INTO METRICS SELECT 'METRIC_'||(LPAD(ENCODE(NEXT VALUE FOR foo.bar,'base62'),5,'0')), user_id, sum(val) FROM events GROUP BY user_id ORDER BY user_id");
+ conn.commit();
+ nextConnection();
+
+ PreparedStatement stmt =
+ conn.prepareStatement("SELECT metric_id, agg_id, metric_val FROM METRICS");
+ ResultSet rs = stmt.executeQuery();
+ assertTrue(rs.next());
+ assertEquals("METRIC_00001", rs.getString("metric_id"));
+ assertEquals("user1", rs.getString("agg_id"));
+ assertEquals(2, rs.getLong("metric_val"));
+ assertTrue(rs.next());
+ assertEquals("METRIC_00002", rs.getString("metric_id"));
+ assertEquals("user2", rs.getString("agg_id"));
+ assertEquals(3, rs.getLong("metric_val"));
+ assertTrue(rs.next());
+ assertEquals("METRIC_00003", rs.getString("metric_id"));
+ assertEquals("user3", rs.getString("agg_id"));
+ assertEquals(1, rs.getLong("metric_val"));
+ assertFalse(rs.next());
+ }
+
+ private void insertEvent(long id, String userId, long val) throws SQLException {
+ PreparedStatement stmt = conn.prepareStatement("UPSERT INTO events VALUES(?,?,?)");
+ stmt.setLong(1, id);
+ stmt.setString(2, userId);
+ stmt.setLong(3, val);
+ stmt.execute();
+ }
+
+ /**
+ * Helper to verify the sequence values returned in multiple ResultSets each containing one row
+ * @param seqVals expected sequence values (one per ResultSet)
+ */
+ private void assertSequenceValuesForSingleRow(long... seqVals)
+ throws SQLException {
+ PreparedStatement stmt = conn.prepareStatement(NEXT_VAL_SQL);
+ for (long seqVal : seqVals) {
+ ResultSet rs = stmt.executeQuery();
+ assertTrue(rs.next());
+ assertEquals(seqVal, rs.getLong(1));
+ assertFalse(rs.next());
+ rs.close();
+ }
+ stmt.close();
+ }
+
+ /**
+ * Helper to verify the sequence values returned in a single ResultSet containing multiple row
+ * @param seqVals expected sequence values (from one ResultSet)
+ */
+ private void assertSequenceValuesMultipleSeq(long... seqVals) throws SQLException {
+ PreparedStatement stmt = conn.prepareStatement(NEXT_VAL_SQL);
+ ResultSet rs = stmt.executeQuery();
+ for (long seqVal : seqVals) {
+ assertTrue(rs.next());
+ assertEquals(seqVal, rs.getLong(1));
+ }
+ assertFalse(rs.next());
+ rs.close();
+ stmt.close();
+ }
+
+ private void verifyExceptions(SQLException sqlE, List<String> expectedExceptions) {
+ List<String> missingExceptions = Lists.newArrayList(expectedExceptions);
+ List<String> unexpectedExceptions = Lists.newArrayList();
+ do {
+ if (!expectedExceptions.contains(sqlE.getMessage())) {
+ unexpectedExceptions.add(sqlE.getMessage());
+ }
+ missingExceptions.remove(sqlE.getMessage());
+ } while ((sqlE = sqlE.getNextException()) != null);
+ if (unexpectedExceptions.size() != 0 && missingExceptions.size() != 0) {
+ fail("Actual exceptions does not match expected exceptions. Unexpected exceptions : "
+ + unexpectedExceptions + " missing exceptions : " + missingExceptions);
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/antlr3/PhoenixSQL.g
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/antlr3/PhoenixSQL.g b/phoenix-core/src/main/antlr3/PhoenixSQL.g
index c4e7f6b..4ba4c99 100644
--- a/phoenix-core/src/main/antlr3/PhoenixSQL.g
+++ b/phoenix-core/src/main/antlr3/PhoenixSQL.g
@@ -101,6 +101,9 @@ tokens
DERIVE='derive';
ANY='any';
SOME='some';
+ MINVALUE='minvalue';
+ MAXVALUE='maxvalue';
+ CYCLE='cycle';
}
@@ -138,6 +141,7 @@ import java.util.Collections;
import java.util.Stack;
import java.sql.SQLException;
import org.apache.phoenix.expression.function.CountAggregateFunction;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.IllegalDataException;
@@ -392,10 +396,13 @@ create_index_node returns [CreateIndexStatement ret]
// Parse a create sequence statement.
create_sequence_node returns [CreateSequenceStatement ret]
: CREATE SEQUENCE (IF NOT ex=EXISTS)? t=from_table_name
- (START WITH? s=int_literal_or_bind)?
- (INCREMENT BY? i=int_literal_or_bind)?
+ (START WITH? s=value_expression)?
+ (INCREMENT BY? i=value_expression)?
+ (MINVALUE min=value_expression)?
+ (MAXVALUE max=value_expression)?
+ (cyc=CYCLE)?
(CACHE c=int_literal_or_bind)?
- { $ret = factory.createSequence(t, s, i, c, ex!=null, getBindCount()); }
+ { $ret = factory.createSequence(t, s, i, c, min, max, cyc!=null, ex!=null, getBindCount()); }
;
int_literal_or_bind returns [ParseNode ret]
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateSequenceCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateSequenceCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateSequenceCompiler.java
index 94f13cb..d31a8f1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateSequenceCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/CreateSequenceCompiler.java
@@ -32,12 +32,14 @@ import org.apache.phoenix.parse.BindParseNode;
import org.apache.phoenix.parse.CreateSequenceStatement;
import org.apache.phoenix.parse.ParseNode;
import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.parse.TableName;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.schema.PDataType;
import org.apache.phoenix.schema.PDatum;
import org.apache.phoenix.schema.SortOrder;
+import org.apache.phoenix.util.SequenceUtil;
public class CreateSequenceCompiler {
private final PhoenixStatement statement;
@@ -105,77 +107,134 @@ public class CreateSequenceCompiler {
private static final PDatum LONG_DATUM = new LongDatum();
private static final PDatum INTEGER_DATUM = new IntegerDatum();
+ private void validateNodeIsStateless(CreateSequenceStatement sequence, ParseNode node,
+ SQLExceptionCode code) throws SQLException {
+ if (!node.isStateless()) {
+ TableName sequenceName = sequence.getSequenceName();
+ throw SequenceUtil.getException(sequenceName.getSchemaName(), sequenceName.getTableName(), code);
+ }
+ }
+
+ private long evalExpression(CreateSequenceStatement sequence, StatementContext context,
+ Expression expression, SQLExceptionCode code) throws SQLException {
+ ImmutableBytesWritable ptr = context.getTempPtr();
+ expression.evaluate(null, ptr);
+ if (ptr.getLength() == 0 || !expression.getDataType().isCoercibleTo(PDataType.LONG)) {
+ TableName sequenceName = sequence.getSequenceName();
+ throw SequenceUtil.getException(sequenceName.getSchemaName(), sequenceName.getTableName(), code);
+ }
+ return (Long) PDataType.LONG.toObject(ptr, expression.getDataType());
+ }
+
public MutationPlan compile(final CreateSequenceStatement sequence) throws SQLException {
ParseNode startsWithNode = sequence.getStartWith();
ParseNode incrementByNode = sequence.getIncrementBy();
- if (!startsWithNode.isStateless()) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.STARTS_WITH_MUST_BE_CONSTANT)
- .setSchemaName(sequence.getSequenceName().getSchemaName())
- .setTableName(sequence.getSequenceName().getTableName()).build().buildException();
- }
- if (!incrementByNode.isStateless()) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.INCREMENT_BY_MUST_BE_CONSTANT)
- .setSchemaName(sequence.getSequenceName().getSchemaName())
- .setTableName(sequence.getSequenceName().getTableName()).build().buildException();
- }
+ ParseNode maxValueNode = sequence.getMaxValue();
+ ParseNode minValueNode = sequence.getMinValue();
ParseNode cacheNode = sequence.getCacheSize();
- if (cacheNode != null && !cacheNode.isStateless()) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.CACHE_MUST_BE_NON_NEGATIVE_CONSTANT)
- .setSchemaName(sequence.getSequenceName().getSchemaName())
- .setTableName(sequence.getSequenceName().getTableName()).build().buildException();
+
+ // validate parse nodes
+ if (startsWithNode!=null) {
+ validateNodeIsStateless(sequence, startsWithNode,
+ SQLExceptionCode.START_WITH_MUST_BE_CONSTANT);
}
-
+ validateNodeIsStateless(sequence, incrementByNode,
+ SQLExceptionCode.INCREMENT_BY_MUST_BE_CONSTANT);
+ validateNodeIsStateless(sequence, maxValueNode,
+ SQLExceptionCode.MAXVALUE_MUST_BE_CONSTANT);
+ validateNodeIsStateless(sequence, minValueNode,
+ SQLExceptionCode.MINVALUE_MUST_BE_CONSTANT);
+ if (cacheNode != null) {
+ validateNodeIsStateless(sequence, cacheNode,
+ SQLExceptionCode.CACHE_MUST_BE_NON_NEGATIVE_CONSTANT);
+ }
+
final PhoenixConnection connection = statement.getConnection();
-
final StatementContext context = new StatementContext(statement);
+
+ // add param meta data if required
if (startsWithNode instanceof BindParseNode) {
- context.getBindManager().addParamMetaData((BindParseNode)startsWithNode, LONG_DATUM);
+ context.getBindManager().addParamMetaData((BindParseNode) startsWithNode, LONG_DATUM);
}
if (incrementByNode instanceof BindParseNode) {
- context.getBindManager().addParamMetaData((BindParseNode)incrementByNode, LONG_DATUM);
+ context.getBindManager().addParamMetaData((BindParseNode) incrementByNode, LONG_DATUM);
+ }
+ if (maxValueNode instanceof BindParseNode) {
+ context.getBindManager().addParamMetaData((BindParseNode) maxValueNode, LONG_DATUM);
+ }
+ if (minValueNode instanceof BindParseNode) {
+ context.getBindManager().addParamMetaData((BindParseNode) minValueNode, LONG_DATUM);
}
if (cacheNode instanceof BindParseNode) {
- context.getBindManager().addParamMetaData((BindParseNode)cacheNode, INTEGER_DATUM);
+ context.getBindManager().addParamMetaData((BindParseNode) cacheNode, INTEGER_DATUM);
}
- ExpressionCompiler expressionCompiler = new ExpressionCompiler(context);
- Expression startsWithExpr = startsWithNode.accept(expressionCompiler);
- ImmutableBytesWritable ptr = context.getTempPtr();
- startsWithExpr.evaluate(null, ptr);
- if (ptr.getLength() == 0 || !startsWithExpr.getDataType().isCoercibleTo(PDataType.LONG)) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.STARTS_WITH_MUST_BE_CONSTANT)
- .setSchemaName(sequence.getSequenceName().getSchemaName())
- .setTableName(sequence.getSequenceName().getTableName()).build().buildException();
- }
- final long startsWith = (Long)PDataType.LONG.toObject(ptr, startsWithExpr.getDataType());
-
- Expression incrementByExpr = incrementByNode.accept(expressionCompiler);
- incrementByExpr.evaluate(null, ptr);
- if (ptr.getLength() == 0 || !incrementByExpr.getDataType().isCoercibleTo(PDataType.LONG)) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.INCREMENT_BY_MUST_BE_CONSTANT)
- .setSchemaName(sequence.getSequenceName().getSchemaName())
- .setTableName(sequence.getSequenceName().getTableName()).build().buildException();
- }
- final long incrementBy = (Long)PDataType.LONG.toObject(ptr, incrementByExpr.getDataType());
- long cacheSizeValue = connection.getQueryServices().getProps().getLong(QueryServices.SEQUENCE_CACHE_SIZE_ATTRIB,QueryServicesOptions.DEFAULT_SEQUENCE_CACHE_SIZE);
- if (cacheNode != null) {
- Expression cacheSizeExpr = cacheNode.accept(expressionCompiler);
- cacheSizeExpr.evaluate(null, ptr);
- if (ptr.getLength() != 0 && (!cacheSizeExpr.getDataType().isCoercibleTo(PDataType.LONG) || (cacheSizeValue = (Long)PDataType.LONG.toObject(ptr, cacheSizeExpr.getDataType())) < 0)) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.CACHE_MUST_BE_NON_NEGATIVE_CONSTANT)
- .setSchemaName(sequence.getSequenceName().getSchemaName())
- .setTableName(sequence.getSequenceName().getTableName()).build().buildException();
+ ExpressionCompiler expressionCompiler = new ExpressionCompiler(context);
+ final long incrementBy =
+ evalExpression(sequence, context, incrementByNode.accept(expressionCompiler),
+ SQLExceptionCode.INCREMENT_BY_MUST_BE_CONSTANT);
+ if (incrementBy == 0) {
+ throw SequenceUtil.getException(sequence.getSequenceName().getSchemaName(), sequence
+ .getSequenceName().getTableName(),
+ SQLExceptionCode.INCREMENT_BY_MUST_NOT_BE_ZERO);
+ }
+ final long maxValue =
+ evalExpression(sequence, context, maxValueNode.accept(expressionCompiler),
+ SQLExceptionCode.MAXVALUE_MUST_BE_CONSTANT);
+ final long minValue =
+ evalExpression(sequence, context, minValueNode.accept(expressionCompiler),
+ SQLExceptionCode.MINVALUE_MUST_BE_CONSTANT);
+ if (minValue>maxValue) {
+ TableName sequenceName = sequence.getSequenceName();
+ throw SequenceUtil.getException(sequenceName.getSchemaName(),
+ sequenceName.getTableName(),
+ SQLExceptionCode.MINVALUE_MUST_BE_LESS_THAN_OR_EQUAL_TO_MAXVALUE);
+ }
+
+ long startsWithValue;
+ if (startsWithNode == null) {
+ startsWithValue = incrementBy > 0 ? minValue : maxValue;
+ } else {
+ startsWithValue =
+ evalExpression(sequence, context, startsWithNode.accept(expressionCompiler),
+ SQLExceptionCode.START_WITH_MUST_BE_CONSTANT);
+ if (startsWithValue < minValue || startsWithValue > maxValue) {
+ TableName sequenceName = sequence.getSequenceName();
+ throw SequenceUtil.getException(sequenceName.getSchemaName(),
+ sequenceName.getTableName(),
+ SQLExceptionCode.STARTS_WITH_MUST_BE_BETWEEN_MIN_MAX_VALUE);
+ }
+ }
+ final long startsWith = startsWithValue;
+
+ long cacheSizeValue;
+ if (cacheNode == null) {
+ cacheSizeValue =
+ connection
+ .getQueryServices()
+ .getProps()
+ .getLong(QueryServices.SEQUENCE_CACHE_SIZE_ATTRIB,
+ QueryServicesOptions.DEFAULT_SEQUENCE_CACHE_SIZE);
+ }
+ else {
+ cacheSizeValue =
+ evalExpression(sequence, context, cacheNode.accept(expressionCompiler),
+ SQLExceptionCode.CACHE_MUST_BE_NON_NEGATIVE_CONSTANT);
+ if (cacheSizeValue < 0) {
+ TableName sequenceName = sequence.getSequenceName();
+ throw SequenceUtil.getException(sequenceName.getSchemaName(),
+ sequenceName.getTableName(),
+ SQLExceptionCode.CACHE_MUST_BE_NON_NEGATIVE_CONSTANT);
}
}
final long cacheSize = Math.max(1L, cacheSizeValue);
-
- final MetaDataClient client = new MetaDataClient(connection);
- return new MutationPlan() {
+ final MetaDataClient client = new MetaDataClient(connection);
+ return new MutationPlan() {
@Override
public MutationState execute() throws SQLException {
- return client.createSequence(sequence, startsWith, incrementBy, cacheSize);
+ return client.createSequence(sequence, startsWith, incrementBy, cacheSize, minValue, maxValue);
}
@Override
@@ -189,7 +248,7 @@ public class CreateSequenceCompiler {
}
@Override
- public ParameterMetaData getParameterMetaData() {
+ public ParameterMetaData getParameterMetaData() {
return context.getBindManager().getParameterMetaData();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
index 875bb0c..cc3c0b4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/SequenceRegionObserver.java
@@ -19,6 +19,7 @@
package org.apache.phoenix.coprocessor;
import java.io.IOException;
+import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -50,6 +51,7 @@ import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.MetaDataUtil;
+import org.apache.phoenix.util.SequenceUtil;
import org.apache.phoenix.util.ServerUtil;
/**
@@ -81,7 +83,6 @@ public class SequenceRegionObserver extends BaseRegionObserver {
QueryConstants.EMPTY_COLUMN_BYTES, timestamp, errorCodeBuf)));
}
/**
- *
* 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
@@ -123,25 +124,112 @@ public class SequenceRegionObserver extends BaseRegionObserver {
if (validateOnly) {
return result;
}
- KeyValue currentValueKV = Sequence.getCurrentValueKV(result);
+
KeyValue incrementByKV = Sequence.getIncrementByKV(result);
- KeyValue cacheSizeKV = Sequence.getCacheSizeKV(result);
- long value = PDataType.LONG.getCodec().decodeLong(currentValueKV.getBuffer(), currentValueKV.getValueOffset(), SortOrder.getDefault());
- long incrementBy = PDataType.LONG.getCodec().decodeLong(incrementByKV.getBuffer(), incrementByKV.getValueOffset(), SortOrder.getDefault());
- long cacheSize = PDataType.LONG.getCodec().decodeLong(cacheSizeKV.getBuffer(), cacheSizeKV.getValueOffset(), SortOrder.getDefault());
- value += incrementBy * cacheSize;
- byte[] valueBuffer = new byte[PDataType.LONG.getByteSize()];
- PDataType.LONG.getCodec().encodeLong(value, valueBuffer, 0);
+ KeyValue currentValueKV = Sequence.getCurrentValueKV(result);
+ KeyValue cacheSizeKV = Sequence.getCacheSizeKV(result);
+ KeyValue cycleKV = Sequence.getCycleKV(result);
+ KeyValue minValueKV = Sequence.getMinValueKV(result);
+ KeyValue maxValueKV = Sequence.getMaxValueKV(result);
+
+ // Hold timestamp constant for sequences, so that clients always only see the latest
+ // value regardless of when they connect.
Put put = new Put(row, currentValueKV.getTimestamp());
- // Hold timestamp constant for sequences, so that clients always only see the latest value
- // regardless of when they connect.
- KeyValue newCurrentValueKV = KeyValueUtil.newKeyValue(row, currentValueKV.getFamily(), currentValueKV.getQualifier(), currentValueKV.getTimestamp(), valueBuffer);
+
+ // create a copy of the key values, used for the new Return
+ List<KeyValue> newkvs = Sequence.getCells(result);
+
+ long incrementBy =
+ PDataType.LONG.getCodec().decodeLong(incrementByKV.getBuffer(),
+ incrementByKV.getValueOffset(), SortOrder.getDefault());
+
+ long cacheSize =
+ PDataType.LONG.getCodec().decodeLong(cacheSizeKV.getBuffer(),
+ cacheSizeKV.getValueOffset(), SortOrder.getDefault());
+
+ // if the minValue, maxValue, or cycle is null this sequence has been upgraded from
+ // a lower version. Set minValue, maxValue and cycle to Long.MIN_VALUE, Long.MAX_VALUE and true
+ // respectively in order to maintain existing behavior and also update the KeyValues on the server
+ long minValue;
+ if (minValueKV == null) {
+ minValue = Long.MIN_VALUE;
+ // create new key value for put
+ byte[] newMinValueBuffer = new byte[PDataType.LONG.getByteSize()];
+ PDataType.LONG.getCodec().encodeLong(minValue, newMinValueBuffer, 0);
+ KeyValue newMinValueKV = KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.MIN_VALUE_BYTES, currentValueKV.getTimestamp(), newMinValueBuffer);
+ put.add(newMinValueKV);
+ // update key value in returned Result
+ Sequence.replaceMinValueKV(newkvs, newMinValueKV);
+ }
+ else {
+ minValue = PDataType.LONG.getCodec().decodeLong(minValueKV.getBuffer(),
+ minValueKV.getValueOffset(), SortOrder.getDefault());
+ }
+ long maxValue;
+ if (maxValueKV == null) {
+ maxValue = Long.MAX_VALUE;
+ // create new key value for put
+ byte[] newMaxValueBuffer = new byte[PDataType.LONG.getByteSize()];
+ PDataType.LONG.getCodec().encodeLong(maxValue, newMaxValueBuffer, 0);
+ KeyValue newMaxValueKV = KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.MAX_VALUE_BYTES, currentValueKV.getTimestamp(), newMaxValueBuffer);
+ put.add(newMaxValueKV);
+ // update key value in returned Result
+ Sequence.replaceMaxValueKV(newkvs, newMaxValueKV);
+ }
+ else {
+ maxValue = PDataType.LONG.getCodec().decodeLong(maxValueKV.getBuffer(),
+ maxValueKV.getValueOffset(), SortOrder.getDefault());
+ }
+ boolean cycle;
+ if (cycleKV == null) {
+ cycle = false;
+ // create new key value for put
+ KeyValue newCycleKV = KeyValueUtil.newKeyValue(row, PhoenixDatabaseMetaData.SEQUENCE_FAMILY_BYTES,
+ PhoenixDatabaseMetaData.CYCLE_FLAG_BYTES, currentValueKV.getTimestamp(), PDataType.FALSE_BYTES);
+ put.add(newCycleKV);
+ // update key value in returned Result
+ Sequence.replaceCycleValueKV(newkvs, newCycleKV);
+ }
+ else {
+ cycle = (Boolean) PDataType.BOOLEAN.toObject(cycleKV.getBuffer(),
+ cycleKV.getValueOffset(), cycleKV.getValueLength());
+ }
+ long currentValue;
+ // initialize current value to start value
+ if (currentValueKV.getValueLength()==0) {
+ KeyValue startValueKV = Sequence.getStartValueKV(result);
+ currentValue =
+ PDataType.LONG.getCodec().decodeLong(startValueKV.getBuffer(),
+ startValueKV.getValueOffset(), SortOrder.getDefault());
+ }
+ else {
+ currentValue =
+ PDataType.LONG.getCodec().decodeLong(currentValueKV.getBuffer(),
+ currentValueKV.getValueOffset(), SortOrder.getDefault());
+ try {
+ // set currentValue to nextValue
+ currentValue =
+ SequenceUtil.getNextValue(currentValue, minValue, maxValue,
+ incrementBy, cacheSize, cycle);
+ } catch (SQLException sqlE) {
+ return getErrorResult(row, maxTimestamp, sqlE.getErrorCode());
+ }
+ }
+ byte[] newCurrentValueBuffer = new byte[PDataType.LONG.getByteSize()];
+ PDataType.LONG.getCodec().encodeLong(currentValue, newCurrentValueBuffer, 0);
+ KeyValue newCurrentValueKV = KeyValueUtil.newKeyValue(row, currentValueKV, newCurrentValueBuffer);
put.add(newCurrentValueKV);
+ Sequence.replaceCurrentValueKV(newkvs, newCurrentValueKV);
+
+ // update the KeyValues on the server
@SuppressWarnings("unchecked")
Pair<Mutation,Integer>[] mutations = new Pair[1];
mutations[0] = new Pair<Mutation,Integer>(put, lid);
region.batchMutate(mutations);
- return Sequence.replaceCurrentValueKV(result, newCurrentValueKV);
+ // return a Result with the updated KeyValues
+ return new Result(newkvs);
} finally {
region.releaseRowLock(lid);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index 39b951d..d6ff6d5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -226,13 +226,21 @@ public enum SQLExceptionCode {
return new SequenceNotFoundException(info.getSchemaName(), info.getTableName());
}
}),
- STARTS_WITH_MUST_BE_CONSTANT(1202, "42Z02", "Sequence STARTS WITH value must be an integer or long constant."),
+ START_WITH_MUST_BE_CONSTANT(1202, "42Z02", "Sequence START WITH value must be an integer or long constant."),
INCREMENT_BY_MUST_BE_CONSTANT(1203, "42Z03", "Sequence INCREMENT BY value must be an integer or long constant."),
CACHE_MUST_BE_NON_NEGATIVE_CONSTANT(1204, "42Z04", "Sequence CACHE value must be a non negative integer constant."),
INVALID_USE_OF_NEXT_VALUE_FOR(1205, "42Z05", "NEXT VALUE FOR may only be used as in a SELECT or an UPSERT VALUES expression."),
CANNOT_CALL_CURRENT_BEFORE_NEXT_VALUE(1206, "42Z06", "NEXT VALUE FOR must be called before CURRENT VALUE FOR is called."),
EMPTY_SEQUENCE_CACHE(1207, "42Z07", "No more cached sequence values"),
-
+ MINVALUE_MUST_BE_CONSTANT(1208, "42Z08", "Sequence MINVALUE must be an integer or long constant."),
+ MAXVALUE_MUST_BE_CONSTANT(1209, "42Z09", "Sequence MAXVALUE must be an integer or long constant."),
+ MINVALUE_MUST_BE_LESS_THAN_OR_EQUAL_TO_MAXVALUE(1210, "42Z10", "Sequence MINVALUE must be less than or equal to MAXVALUE."),
+ STARTS_WITH_MUST_BE_BETWEEN_MIN_MAX_VALUE(1211, "42Z11",
+ "STARTS WITH value must be greater than or equal to MINVALUE and less than or equal to MAXVALUE"),
+ SEQUENCE_VAL_REACHED_MAX_VALUE(1212, "42Z12", "Reached MAXVALUE of sequence"),
+ SEQUENCE_VAL_REACHED_MIN_VALUE(1213, "42Z13", "Reached MINVALUE of sequence"),
+ INCREMENT_BY_MUST_NOT_BE_ZERO(1214, "42Z14", "Sequence INCREMENT BY value cannot be zero"),
+
/** Parser error. (errorcode 06, sqlState 42P) */
PARSER_ERROR(601, "42P00", "Syntax error.", Factory.SYTAX_ERROR),
MISSING_TOKEN(602, "42P00", "Syntax error.", Factory.SYTAX_ERROR),
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index a41935f..27a53e4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -198,10 +198,17 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
public static final byte[] CURRENT_VALUE_BYTES = Bytes.toBytes(CURRENT_VALUE);
public static final String START_WITH = "START_WITH";
public static final byte[] START_WITH_BYTES = Bytes.toBytes(START_WITH);
+ // MIN_VALUE, MAX_VALUE and CYCLE_FLAG were added in 3.0
+ public static final String MIN_VALUE = "MIN_VALUE";
+ public static final byte[] MIN_VALUE_BYTES = Bytes.toBytes(MIN_VALUE);
+ public static final String MAX_VALUE = "MAX_VALUE";
+ public static final byte[] MAX_VALUE_BYTES = Bytes.toBytes(MAX_VALUE);
public static final String INCREMENT_BY = "INCREMENT_BY";
public static final byte[] INCREMENT_BY_BYTES = Bytes.toBytes(INCREMENT_BY);
public static final String CACHE_SIZE = "CACHE_SIZE";
public static final byte[] CACHE_SIZE_BYTES = Bytes.toBytes(CACHE_SIZE);
+ public static final String CYCLE_FLAG = "CYCLE_FLAG";
+ public static final byte[] CYCLE_FLAG_BYTES = Bytes.toBytes(CYCLE_FLAG);
public static final String KEY_SEQ = "KEY_SEQ";
public static final byte[] KEY_SEQ_BYTES = Bytes.toBytes(KEY_SEQ);
public static final String SUPERTABLE_NAME = "SUPERTABLE_NAME";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index d4c677b..4b63cfe 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -487,9 +487,12 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
private static class ExecutableCreateSequenceStatement extends CreateSequenceStatement implements CompilableStatement {
- public ExecutableCreateSequenceStatement(TableName sequenceName, ParseNode startWith, ParseNode incrementBy, ParseNode cacheSize, boolean ifNotExists, int bindCount) {
- super(sequenceName, startWith, incrementBy, cacheSize, ifNotExists, bindCount);
- }
+ public ExecutableCreateSequenceStatement(TableName sequenceName, ParseNode startWith,
+ ParseNode incrementBy, ParseNode cacheSize, ParseNode minValue, ParseNode maxValue,
+ boolean cycle, boolean ifNotExists, int bindCount) {
+ super(sequenceName, startWith, incrementBy, cacheSize, minValue, maxValue, cycle,
+ ifNotExists, bindCount);
+ }
@SuppressWarnings("unchecked")
@Override
@@ -744,8 +747,11 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
}
@Override
- public CreateSequenceStatement createSequence(TableName tableName, ParseNode startsWith, ParseNode incrementBy, ParseNode cacheSize, boolean ifNotExists, int bindCount){
- return new ExecutableCreateSequenceStatement(tableName, startsWith, incrementBy, cacheSize, ifNotExists, bindCount);
+ public CreateSequenceStatement createSequence(TableName tableName, ParseNode startsWith,
+ ParseNode incrementBy, ParseNode cacheSize, ParseNode minValue, ParseNode maxValue,
+ boolean cycle, boolean ifNotExists, int bindCount) {
+ return new ExecutableCreateSequenceStatement(tableName, startsWith, incrementBy,
+ cacheSize, minValue, maxValue, cycle, ifNotExists, bindCount);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateSequenceStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateSequenceStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateSequenceStatement.java
index 44d5f34..2e0c943 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateSequenceStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateSequenceStatement.java
@@ -20,24 +20,37 @@ package org.apache.phoenix.parse;
public class CreateSequenceStatement extends MutableStatement {
public static CreateSequenceStatement create(TableName sequenceName) {
- return new CreateSequenceStatement(sequenceName, null, null, null, true, 0);
+ return new CreateSequenceStatement(sequenceName, null, null, null, null, null, false, true,
+ 0);
}
private final TableName sequenceName;
private final ParseNode startWith;
private final ParseNode incrementBy;
private final ParseNode cacheSize;
+ private final ParseNode minValue;
+ private final ParseNode maxValue;
+ private final boolean cycle;
private final boolean ifNotExists;
private final int bindCount;
- protected CreateSequenceStatement(TableName sequenceName, ParseNode startsWith, ParseNode incrementBy, ParseNode cacheSize, boolean ifNotExists, int bindCount) {
- this.sequenceName = sequenceName;
- this.startWith = startsWith == null ? LiteralParseNode.ONE : startsWith;
- this.incrementBy = incrementBy == null ? LiteralParseNode.ONE : incrementBy;
- this.cacheSize = cacheSize == null ? null : cacheSize;
- this.ifNotExists = ifNotExists;
- this.bindCount = bindCount;
- }
+ protected CreateSequenceStatement(TableName sequenceName, ParseNode startWith,
+ ParseNode incrementBy, ParseNode cacheSize, ParseNode minValue, ParseNode maxValue,
+ boolean cycle, boolean ifNotExists, int bindCount) {
+ this.sequenceName = sequenceName;
+ // if MINVALUE, MAXVALUE and START WITH are not specified, set START WITH to 1 in order to
+ // maintain backward compatibility
+ this.startWith =
+ (minValue == null && maxValue == null && startWith == null) ? LiteralParseNode.ONE
+ : startWith;
+ this.minValue = minValue == null ? new LiteralParseNode(Long.MIN_VALUE) : minValue;
+ this.maxValue = maxValue == null ? new LiteralParseNode(Long.MAX_VALUE) : maxValue;
+ this.incrementBy = incrementBy == null ? LiteralParseNode.ONE : incrementBy;
+ this.cacheSize = cacheSize;
+ this.cycle = cycle;
+ this.ifNotExists = ifNotExists;
+ this.bindCount = bindCount;
+ }
@Override
public int getBindCount() {
@@ -56,7 +69,19 @@ public class CreateSequenceStatement extends MutableStatement {
return cacheSize;
}
- public ParseNode getStartWith() {
+ public ParseNode getMinValue() {
+ return minValue;
+ }
+
+ public ParseNode getMaxValue() {
+ return maxValue;
+ }
+
+ public boolean getCycle() {
+ return cycle;
+ }
+
+ public ParseNode getStartWith() {
return startWith;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
index 4b27696..088902e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
@@ -272,9 +272,12 @@ public class ParseNodeFactory {
return new CreateIndexStatement(indexName, dataTable, pkConstraint, includeColumns, splits, props, ifNotExists, bindCount);
}
- public CreateSequenceStatement createSequence(TableName tableName, ParseNode startsWith, ParseNode incrementBy, ParseNode cacheSize, boolean ifNotExits, int bindCount){
- return new CreateSequenceStatement(tableName, startsWith, incrementBy, cacheSize, ifNotExits, bindCount);
- }
+ public CreateSequenceStatement createSequence(TableName tableName, ParseNode startsWith,
+ ParseNode incrementBy, ParseNode cacheSize, ParseNode minValue, ParseNode maxValue,
+ boolean cycle, boolean ifNotExits, int bindCount) {
+ return new CreateSequenceStatement(tableName, startsWith, incrementBy, cacheSize, minValue,
+ maxValue, cycle, ifNotExits, bindCount);
+ }
public DropSequenceStatement dropSequence(TableName tableName, boolean ifExits, int bindCount){
return new DropSequenceStatement(tableName, ifExits, bindCount);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
index 055bc18..8270710 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
@@ -37,7 +37,7 @@ public class SelectStatement implements FilterableStatement {
public static final SelectStatement SELECT_ONE =
new SelectStatement(
Collections.<TableNode>emptyList(), null, false,
- Collections.<AliasedNode>singletonList(new AliasedNode(null,new LiteralParseNode(1))),
+ Collections.<AliasedNode>singletonList(new AliasedNode(null, LiteralParseNode.ONE)),
null, Collections.<ParseNode>emptyList(),
null, Collections.<OrderByNode>emptyList(),
null, 0, false);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
index 035c77b..f085464 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
@@ -86,7 +86,7 @@ public interface ConnectionQueryServices extends QueryServices, MetaDataMutated
boolean hasInvalidIndexConfiguration();
- long createSequence(String tenantId, String schemaName, String sequenceName, long startWith, long incrementBy, long cacheSize, long timestamp) throws SQLException;
+ long createSequence(String tenantId, String schemaName, String sequenceName, long startWith, long incrementBy, long cacheSize, long minValue, long maxValue, boolean cycle, long timestamp) throws SQLException;
long dropSequence(String tenantId, String schemaName, String sequenceName, long timestamp) throws SQLException;
void validateSequences(List<SequenceKey> sequenceKeys, long timestamp, long[] values, SQLException[] exceptions, Sequence.ValueOp action) throws SQLException;
void incrementSequences(List<SequenceKey> sequenceKeys, long timestamp, long[] values, SQLException[] exceptions) throws SQLException;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 21b998b..d02859b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -22,6 +22,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_COUNT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_FAMILY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_SIZE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CYCLE_FLAG;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TABLE_NAME_BYTES;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TYPE;
@@ -31,6 +32,8 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_ROWS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_STATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.KEY_SEQ;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LINK_TYPE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MAX_VALUE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MIN_VALUE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NULLABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ORDINAL_POSITION;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PK_NAME;
@@ -189,7 +192,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
private volatile boolean initialized;
private volatile boolean closed;
private volatile SQLException initializationException;
- private ConcurrentMap<SequenceKey,Sequence> sequenceMap = Maps.newConcurrentMap();
+ protected ConcurrentMap<SequenceKey,Sequence> sequenceMap = Maps.newConcurrentMap();
private KeyValueBuilder kvBuilder;
private PMetaData newEmptyMetaData() {
@@ -1370,7 +1373,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
try {
metaConnection.createStatement().executeUpdate(QueryConstants.CREATE_TABLE_METADATA);
} catch (NewerTableAlreadyExistsException ignore) {
- // Ignore, as this will happen if the SYSTEM.TABLE already exists at this fixed timestamp.
+ // Ignore, as this will happen if the SYSTEM.CATALOG already exists at this fixed timestamp.
// A TableAlreadyExistsException is not thrown, since the table only exists *after* this fixed timestamp.
} catch (TableAlreadyExistsException ignore) {
// This will occur if we have an older SYSTEM.CATALOG and we need to update it to include
@@ -1382,6 +1385,15 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
} catch (NewerTableAlreadyExistsException ignore) {
// Ignore, as this will happen if the SYSTEM.SEQUENCE already exists at this fixed timestamp.
// A TableAlreadyExistsException is not thrown, since the table only exists *after* this fixed timestamp.
+ } catch (TableAlreadyExistsException ignore) {
+ // This will occur if we have an older SYSTEM.SEQUENCE, so we need to update it to include
+ // any new columns we've added.
+ String newColumns =
+ MIN_VALUE + " " + PDataType.LONG.getSqlTypeName() + ", "
+ + MAX_VALUE + " " + PDataType.LONG.getSqlTypeName() + ", "
+ + CYCLE_FLAG + " " + PDataType.BOOLEAN.getSqlTypeName();
+ metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SEQUENCE_TABLE_NAME,
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP, newColumns);
}
upgradeMetaDataTo3_0(url, props);
} catch (SQLException e) {
@@ -1510,8 +1522,9 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
}
@Override
- public long createSequence(String tenantId, String schemaName, String sequenceName, long startWith, long incrementBy, long cacheSize, long timestamp)
- throws SQLException {
+ public long createSequence(String tenantId, String schemaName, String sequenceName,
+ long startWith, long incrementBy, long cacheSize, long minValue, long maxValue,
+ boolean cycle, long timestamp) throws SQLException {
SequenceKey sequenceKey = new SequenceKey(tenantId, schemaName, sequenceName);
Sequence newSequences = new Sequence(sequenceKey);
Sequence sequence = sequenceMap.putIfAbsent(sequenceKey, newSequences);
@@ -1521,11 +1534,12 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
try {
sequence.getLock().lock();
// Now that we have the lock we need, create the sequence
- Append append = sequence.createSequence(startWith, incrementBy, cacheSize, timestamp);
- HTableInterface htable = this.getTable(PhoenixDatabaseMetaData.SEQUENCE_TABLE_NAME_BYTES);
+ Append append = sequence.createSequence(startWith, incrementBy, cacheSize, timestamp, minValue, maxValue, cycle);
+ HTableInterface htable =
+ this.getTable(PhoenixDatabaseMetaData.SEQUENCE_TABLE_NAME_BYTES);
try {
Result result = htable.append(append);
- return sequence.createSequence(result);
+ return sequence.createSequence(result, minValue, maxValue, cycle);
} catch (IOException e) {
throw ServerUtil.parseServerException(e);
} finally {
@@ -1639,6 +1653,8 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
toIncrementList.add(sequence);
Increment inc = sequence.newIncrement(timestamp, action);
incrementBatch.add(inc);
+ } catch (SQLException e) {
+ exceptions[i] = e;
}
}
if (toIncrementList.isEmpty()) {
@@ -1757,6 +1773,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
// Take no locks, as this only gets run when there are no open connections
// so there's no danger of contention.
+ @SuppressWarnings("deprecation")
private void returnAllSequences(ConcurrentMap<SequenceKey,Sequence> sequenceMap) throws SQLException {
List<Append> mutations = Lists.newArrayListWithExpectedSize(sequenceMap.size());
for (Sequence sequence : sequenceMap.values()) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index d264d38..a48f811 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@ -59,6 +59,7 @@ import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.Sequence;
import org.apache.phoenix.schema.SequenceAlreadyExistsException;
+import org.apache.phoenix.schema.SequenceInfo;
import org.apache.phoenix.schema.SequenceKey;
import org.apache.phoenix.schema.SequenceNotFoundException;
import org.apache.phoenix.schema.TableAlreadyExistsException;
@@ -68,6 +69,7 @@ import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.SequenceUtil;
import com.google.common.collect.Maps;
@@ -82,7 +84,7 @@ import com.google.common.collect.Maps;
*/
public class ConnectionlessQueryServicesImpl extends DelegateQueryServices implements ConnectionQueryServices {
private PMetaData metaData;
- private final Map<SequenceKey, Long> sequenceMap = Maps.newHashMap();
+ private final Map<SequenceKey, SequenceInfo> sequenceMap = Maps.newHashMap();
private KeyValueBuilder kvBuilder;
private volatile boolean initialized;
private volatile SQLException initializationException;
@@ -312,13 +314,14 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
}
@Override
- public long createSequence(String tenantId, String schemaName, String sequenceName, long startWith, long incrementBy, long cacheSize, long timestamp)
- throws SQLException {
+ public long createSequence(String tenantId, String schemaName, String sequenceName,
+ long startWith, long incrementBy, long cacheSize, long minValue, long maxValue,
+ boolean cycle, long timestamp) throws SQLException {
SequenceKey key = new SequenceKey(tenantId, schemaName, sequenceName);
if (sequenceMap.get(key) != null) {
throw new SequenceAlreadyExistsException(schemaName, sequenceName);
}
- sequenceMap.put(key, startWith);
+ sequenceMap.put(key, new SequenceInfo(startWith, incrementBy, minValue, maxValue, 1l, cycle)) ;
return timestamp;
}
@@ -336,11 +339,11 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
SQLException[] exceptions, Sequence.ValueOp action) throws SQLException {
int i = 0;
for (SequenceKey key : sequenceKeys) {
- Long value = sequenceMap.get(key);
- if (value == null) {
+ SequenceInfo info = sequenceMap.get(key);
+ if (info == null) {
exceptions[i] = new SequenceNotFoundException(key.getSchemaName(), key.getSequenceName());
} else {
- values[i] = value;
+ values[i] = info.sequenceValue;
}
i++;
}
@@ -351,11 +354,14 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
SQLException[] exceptions) throws SQLException {
int i = 0;
for (SequenceKey key : sequenceKeys) {
- Long value = sequenceMap.get(key);
- if (value == null) {
- exceptions[i] = new SequenceNotFoundException(key.getSchemaName(), key.getSequenceName());
+ SequenceInfo info = sequenceMap.get(key);
+ if (info == null) {
+ exceptions[i] =
+ new SequenceNotFoundException(key.getSchemaName(), key.getSequenceName());
} else {
- values[i] = value++;
+ values[i] = info.sequenceValue;
+ info.sequenceValue =
+ SequenceUtil.getNextValue(key, info);
}
i++;
}
@@ -370,13 +376,13 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
@Override
public long currentSequenceValue(SequenceKey sequenceKey, long timestamp) throws SQLException {
- Long value = sequenceMap.get(sequenceKey);
- if (value == null) {
+ SequenceInfo info = sequenceMap.get(sequenceKey);
+ if (info == null) {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_CALL_CURRENT_BEFORE_NEXT_VALUE)
.setSchemaName(sequenceKey.getSchemaName()).setTableName(sequenceKey.getSequenceName())
.build().buildException();
}
- return value;
+ return info.sequenceValue;
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/15e1108f/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
index cae52dd..4fa7993 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
@@ -171,9 +171,11 @@ public class DelegateConnectionQueryServices extends DelegateQueryServices imple
}
@Override
- public long createSequence(String tenantId, String schemaName, String sequenceName, long startWith,
- long incrementBy, long cacheSize, long timestamp) throws SQLException {
- return getDelegate().createSequence(tenantId, schemaName, sequenceName, startWith, incrementBy, cacheSize, timestamp);
+ public long createSequence(String tenantId, String schemaName, String sequenceName,
+ long startWith, long incrementBy, long cacheSize, long minValue, long maxValue,
+ boolean cycle, long timestamp) throws SQLException {
+ return getDelegate().createSequence(tenantId, schemaName, sequenceName, startWith,
+ incrementBy, cacheSize, minValue, maxValue, cycle, timestamp);
}
@Override
[3/3] git commit: PHOENIX-1100 Upsert Select with Sequence and Order
By doesn't result in sequence getting assigned based on requested ordering
Posted by ja...@apache.org.
PHOENIX-1100 Upsert Select with Sequence and Order By doesn't result in sequence getting assigned based on requested ordering
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/866d4326
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/866d4326
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/866d4326
Branch: refs/heads/3.0
Commit: 866d43266b41b85ec64d701ac98bb49f0bd49fd5
Parents: 15e1108
Author: James Taylor <jt...@salesforce.com>
Authored: Sat Jul 19 16:43:53 2014 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Sat Jul 19 16:43:53 2014 -0700
----------------------------------------------------------------------
.../apache/phoenix/end2end/UpsertSelectIT.java | 57 ++++++++++++++++++++
phoenix-core/src/main/antlr3/PhoenixSQL.g | 6 ++-
.../apache/phoenix/compile/DeleteCompiler.java | 2 +-
.../apache/phoenix/compile/JoinCompiler.java | 12 +++--
.../phoenix/compile/StatementNormalizer.java | 2 +-
.../phoenix/compile/SubselectRewriter.java | 8 +--
.../apache/phoenix/compile/UpsertCompiler.java | 12 ++---
.../apache/phoenix/jdbc/PhoenixStatement.java | 9 ++--
.../org/apache/phoenix/parse/ParseContext.java | 9 ++++
.../apache/phoenix/parse/ParseNodeFactory.java | 16 ++++--
.../apache/phoenix/parse/ParseNodeRewriter.java | 2 +-
.../apache/phoenix/parse/SelectStatement.java | 20 ++++---
12 files changed, 121 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
index ed79fcc..d3a9ef6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
@@ -796,4 +796,61 @@ public class UpsertSelectIT extends BaseClientManagedTimeIT {
conn.close();
}
+
+ @Test
+ public void testUpsertSelectWithSequenceAndOrderByWithSalting() throws Exception {
+
+ int numOfRecords = 200;
+ long ts = nextTimestamp();
+ Properties props = new Properties(TestUtil.TEST_PROPERTIES);
+ props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts));
+ Connection conn = DriverManager.getConnection(getUrl(), props);
+ String ddl = "CREATE TABLE IF NOT EXISTS DUMMY_CURSOR_STORAGE ("
+ + "ORGANIZATION_ID CHAR(15) NOT NULL, QUERY_ID CHAR(15) NOT NULL, CURSOR_ORDER BIGINT NOT NULL, K1 INTEGER, V1 INTEGER "
+ + "CONSTRAINT MAIN_PK PRIMARY KEY (ORGANIZATION_ID, QUERY_ID, CURSOR_ORDER) " + ") SALT_BUCKETS = 4";
+ conn.createStatement().execute(ddl);
+ conn.createStatement().execute(
+ "CREATE TABLE DUMMY_SEQ_TEST_DATA "
+ + "(ORGANIZATION_ID CHAR(15) NOT NULL, k1 integer NOT NULL, v1 integer NOT NULL "
+ + "CONSTRAINT PK PRIMARY KEY (ORGANIZATION_ID, k1, v1) ) VERSIONS=1, SALT_BUCKETS = 4");
+ conn.createStatement().execute("create sequence s cache " + Integer.MAX_VALUE);
+ conn.close();
+
+ props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts + 10));
+ conn = DriverManager.getConnection(getUrl(), props);
+ for (int i = 0; i < numOfRecords; i++) {
+ conn.createStatement().execute(
+ "UPSERT INTO DUMMY_SEQ_TEST_DATA values ('00Dxx0000001gEH'," + i + "," + (i + 2) + ")");
+ }
+ conn.commit();
+ conn.close();
+
+ props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts + 15));
+ conn = DriverManager.getConnection(getUrl(), props);
+ conn.setAutoCommit(true);
+ conn.createStatement().execute(
+ "UPSERT INTO DUMMY_CURSOR_STORAGE SELECT '00Dxx0000001gEH', 'MyQueryId', NEXT VALUE FOR S, k1, v1 FROM DUMMY_SEQ_TEST_DATA ORDER BY K1, V1");
+
+ props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts + 20));
+ conn = DriverManager.getConnection(getUrl(), props);
+ ResultSet rs = conn.createStatement().executeQuery("select count(*) from DUMMY_CURSOR_STORAGE");
+
+ assertTrue(rs.next());
+ assertEquals(numOfRecords, rs.getLong(1));
+ conn.close();
+
+ props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts + 25));
+ ResultSet rs2 = conn.createStatement().executeQuery(
+ "select cursor_order, k1, v1 from DUMMY_CURSOR_STORAGE order by cursor_order");
+ long seq = 1;
+ while (rs2.next()) {
+ assertEquals(seq, rs2.getLong("cursor_order"));
+ // This value should be the sequence - 1 as we said order by k1 in the UPSERT...SELECT, but is not because
+ // of sequence processing.
+ assertEquals(seq - 1, rs2.getLong("k1"));
+ seq++;
+ }
+ conn.close();
+
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/antlr3/PhoenixSQL.g
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/antlr3/PhoenixSQL.g b/phoenix-core/src/main/antlr3/PhoenixSQL.g
index 4ba4c99..d1b69b8 100644
--- a/phoenix-core/src/main/antlr3/PhoenixSQL.g
+++ b/phoenix-core/src/main/antlr3/PhoenixSQL.g
@@ -556,7 +556,7 @@ select_node returns [SelectStatement ret]
(HAVING having=expression)?
(ORDER BY order=order_by)?
(LIMIT l=limit)?
- { ParseContext context = contextStack.pop(); $ret = factory.select(from, null, d!=null, sel, where, group, having, order, l, getBindCount(), context.isAggregate()); }
+ { ParseContext context = contextStack.pop(); $ret = factory.select(from, null, d!=null, sel, where, group, having, order, l, getBindCount(), context.isAggregate(), context.hasSequences()); }
;
// Parse a full select expression structure.
@@ -806,7 +806,9 @@ term returns [ParseNode ret]
scale == null ? null : Integer.parseInt(scale.getText()),
ar!=null);
}
- | (n=NEXT | CURRENT) VALUE FOR s=from_table_name { $ret = n==null ? factory.currentValueFor(s) : factory.nextValueFor(s);}
+ | (n=NEXT | CURRENT) VALUE FOR s=from_table_name
+ { contextStack.peek().hasSequences(true);
+ $ret = n==null ? factory.currentValueFor(s) : factory.nextValueFor(s); }
;
one_or_more_expressions returns [List<ParseNode> ret]
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
index 1f0eef3..002f7b8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/DeleteCompiler.java
@@ -208,7 +208,7 @@ public class DeleteCompiler {
hint, false, aliasedNodes, delete.getWhere(),
Collections.<ParseNode>emptyList(), null,
delete.getOrderBy(), delete.getLimit(),
- delete.getBindCount(), false);
+ delete.getBindCount(), false, false);
select = StatementNormalizer.normalize(select, resolver);
DeletingParallelIteratorFactory parallelIteratorFactory = hasLimit ? null : new DeletingParallelIteratorFactory(connection, tableRef);
final QueryPlan plan = new QueryOptimizer(services).optimize(statement, select, resolver, Collections.<PColumn>emptyList(), parallelIteratorFactory);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
index 50acaf1..51e2b68 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
@@ -402,7 +402,7 @@ public class JoinCompiler {
if (asSubquery)
return query;
- return NODE_FACTORY.select(query.getFrom(), select.getHint(), select.isDistinct(), select.getSelect(), query.getWhere(), select.getGroupBy(), select.getHaving(), select.getOrderBy(), select.getLimit(), select.getBindCount(), select.isAggregate());
+ return NODE_FACTORY.select(query.getFrom(), select.getHint(), select.isDistinct(), select.getSelect(), query.getWhere(), select.getGroupBy(), select.getHaving(), select.getOrderBy(), select.getLimit(), select.getBindCount(), select.isAggregate(), select.hasSequence());
}
public boolean hasPostReference() {
@@ -660,7 +660,7 @@ public class JoinCompiler {
return SubselectRewriter.applyPostFilters(subselect, postFilters, tableNode.getAlias());
List<TableNode> from = Collections.<TableNode>singletonList(tableNode);
- return NODE_FACTORY.select(from, select.getHint(), false, selectNodes, getPreFiltersCombined(), null, null, null, null, 0, false);
+ return NODE_FACTORY.select(from, select.getHint(), false, selectNodes, getPreFiltersCombined(), null, null, null, null, 0, false, select.hasSequence());
}
public boolean isFlat() {
@@ -1134,7 +1134,7 @@ public class JoinCompiler {
TableRef tableRef = table.getTableRef();
List<ParseNode> groupBy = tableRef.equals(groupByTableRef) ? select.getGroupBy() : null;
List<OrderByNode> orderBy = tableRef.equals(orderByTableRef) ? select.getOrderBy() : null;
- SelectStatement stmt = getSubqueryForOptimizedPlan(select.getHint(), table.getDynamicColumns(), tableRef, join.getColumnRefs(), table.getPreFiltersCombined(), groupBy, orderBy, table.isWildCardSelect());
+ SelectStatement stmt = getSubqueryForOptimizedPlan(select.getHint(), table.getDynamicColumns(), tableRef, join.getColumnRefs(), table.getPreFiltersCombined(), groupBy, orderBy, table.isWildCardSelect(), select.hasSequence());
QueryPlan plan = statement.getConnection().getQueryServices().getOptimizer().optimize(statement, stmt);
if (!plan.getTableRef().equals(tableRef)) {
replacement.put(tableRef, plan.getTableRef());
@@ -1206,7 +1206,8 @@ public class JoinCompiler {
return IndexStatementRewriter.translate(NODE_FACTORY.select(select, newFrom), resolver, replacement);
}
- private static SelectStatement getSubqueryForOptimizedPlan(HintNode hintNode, List<ColumnDef> dynamicCols, TableRef tableRef, Map<ColumnRef, ColumnRefType> columnRefs, ParseNode where, List<ParseNode> groupBy, List<OrderByNode> orderBy, boolean isWildCardSelect) {
+ private static SelectStatement getSubqueryForOptimizedPlan(HintNode hintNode, List<ColumnDef> dynamicCols, TableRef tableRef, Map<ColumnRef, ColumnRefType> columnRefs, ParseNode where, List<ParseNode> groupBy,
+ List<OrderByNode> orderBy, boolean isWildCardSelect, boolean hasSequence) {
String schemaName = tableRef.getTable().getSchemaName().getString();
TableName tName = TableName.create(schemaName.length() == 0 ? null : schemaName, tableRef.getTable().getTableName().getString());
List<AliasedNode> selectList = new ArrayList<AliasedNode>();
@@ -1226,7 +1227,8 @@ public class JoinCompiler {
String tableAlias = tableRef.getTableAlias();
List<? extends TableNode> from = Collections.singletonList(NODE_FACTORY.namedTable(tableAlias == null ? null : '"' + tableAlias + '"', tName, dynamicCols));
- return NODE_FACTORY.select(from, hintNode, false, selectList, where, groupBy, null, orderBy, null, 0, false);
+ // TODO: review, as it seems like we're potentially losing if the select statement is an aggregate (i.e. if it's an ungrouped aggregate for example)
+ return NODE_FACTORY.select(from, hintNode, false, selectList, where, groupBy, null, orderBy, null, 0, groupBy != null, hasSequence);
}
public class PTableWrapper {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementNormalizer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementNormalizer.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementNormalizer.java
index e02f7c3..e29a20a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementNormalizer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementNormalizer.java
@@ -101,7 +101,7 @@ public class StatementNormalizer extends ParseNodeRewriter {
if (selectNodes != normSelectNodes) {
statement = NODE_FACTORY.select(statement.getFrom(), statement.getHint(), statement.isDistinct(),
normSelectNodes, statement.getWhere(), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(),
- statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ statement.getLimit(), statement.getBindCount(), statement.isAggregate(), statement.hasSequence());
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/compile/SubselectRewriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/SubselectRewriter.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/SubselectRewriter.java
index a3d9ebb..3edcbc2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/SubselectRewriter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/SubselectRewriter.java
@@ -196,7 +196,7 @@ public class SubselectRewriter extends ParseNodeRewriter {
isAggregateRewrite = true;
}
- return NODE_FACTORY.select(subselect.getFrom(), hintRewrite, isDistinctRewrite, selectNodesRewrite, whereRewrite, groupByRewrite, havingRewrite, orderByRewrite, limitRewrite, select.getBindCount(), isAggregateRewrite);
+ return NODE_FACTORY.select(subselect.getFrom(), hintRewrite, isDistinctRewrite, selectNodesRewrite, whereRewrite, groupByRewrite, havingRewrite, orderByRewrite, limitRewrite, select.getBindCount(), isAggregateRewrite, select.hasSequence());
}
private SelectStatement applyPostFilters(SelectStatement statement, List<ParseNode> postFilters) throws SQLException {
@@ -210,14 +210,16 @@ public class SubselectRewriter extends ParseNodeRewriter {
if (where != null) {
postFiltersRewrite.add(where);
}
- return NODE_FACTORY.select(statement.getFrom(), statement.getHint(), statement.isDistinct(), statement.getSelect(), combine(postFiltersRewrite), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ return NODE_FACTORY.select(statement.getFrom(), statement.getHint(), statement.isDistinct(), statement.getSelect(), combine(postFiltersRewrite), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(), statement.getLimit(),
+ statement.getBindCount(), statement.isAggregate(), statement.hasSequence());
}
ParseNode having = statement.getHaving();
if (having != null) {
postFiltersRewrite.add(having);
}
- return NODE_FACTORY.select(statement.getFrom(), statement.getHint(), statement.isDistinct(), statement.getSelect(), statement.getWhere(), statement.getGroupBy(), combine(postFiltersRewrite), statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ return NODE_FACTORY.select(statement.getFrom(), statement.getHint(), statement.isDistinct(), statement.getSelect(), statement.getWhere(), statement.getGroupBy(), combine(postFiltersRewrite), statement.getOrderBy(), statement.getLimit(),
+ statement.getBindCount(), statement.isAggregate(), statement.hasSequence());
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
index 60e7bb8..d0d53ab 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
@@ -49,7 +49,6 @@ import org.apache.phoenix.index.IndexMetaDataCacheClient;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.iterate.ParallelIterators.ParallelIteratorFactory;
import org.apache.phoenix.iterate.ResultIterator;
-import org.apache.phoenix.iterate.SequenceResultIterator;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.jdbc.PhoenixStatement;
@@ -181,9 +180,8 @@ public class UpsertCompiler {
@Override
protected MutationState mutate(StatementContext context, ResultIterator iterator, PhoenixConnection connection) throws SQLException {
PhoenixStatement statement = new PhoenixStatement(connection);
- // Clone the connection as it's not thread safe and will be operated on in parallel
if (context.getSequenceManager().getSequenceCount() > 0) {
- iterator = new SequenceResultIterator(iterator,context.getSequenceManager());
+ throw new IllegalStateException("Cannot pipeline upsert when sequence is referenced");
}
return upsertSelect(statement, tableRef, projector, iterator, columnIndexes, pkSlotIndexes);
}
@@ -360,17 +358,19 @@ public class UpsertCompiler {
&& tableRef.equals(selectResolver.getTables().get(0));
/* We can run the upsert in a coprocessor if:
* 1) from has only 1 table and the into table matches from table
- * 2) the select query isn't doing aggregation
+ * 2) the select query isn't doing aggregation (which requires a client-side final merge)
* 3) autoCommit is on
* 4) the table is not immutable, as the client is the one that figures out the additional
* puts for index tables.
- * 5) no limit clause
+ * 5) no limit clause, as the limit clause requires client-side post processing
+ * 6) no sequences, as sequences imply that the order of upsert must match the order of
+ * selection.
* Otherwise, run the query to pull the data from the server
* and populate the MutationState (upto a limit).
*/
runOnServer = sameTable && isAutoCommit && !table.isImmutableRows() && !select.isAggregate() && !select.isDistinct() && select.getLimit() == null && table.getBucketNum() == null;
ParallelIteratorFactory parallelIteratorFactory;
- if (select.isAggregate() || select.isDistinct() || select.getLimit() != null) {
+ if (select.isAggregate() || select.isDistinct() || select.getLimit() != null || select.hasSequence()) {
parallelIteratorFactory = null;
} else {
// We can pipeline the upsert select instead of spooling everything to disk first,
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index 4b63cfe..62e1cc2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -278,8 +278,8 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
private static class ExecutableSelectStatement extends SelectStatement implements CompilableStatement {
private ExecutableSelectStatement(List<? extends TableNode> from, HintNode hint, boolean isDistinct, List<AliasedNode> select, ParseNode where,
- List<ParseNode> groupBy, ParseNode having, List<OrderByNode> orderBy, LimitNode limit, int bindCount, boolean isAggregate) {
- super(from, hint, isDistinct, select, where, groupBy, having, orderBy, limit, bindCount, isAggregate);
+ List<ParseNode> groupBy, ParseNode having, List<OrderByNode> orderBy, LimitNode limit, int bindCount, boolean isAggregate, boolean hasSequence) {
+ super(from, hint, isDistinct, select, where, groupBy, having, orderBy, limit, bindCount, isAggregate, hasSequence);
}
@SuppressWarnings("unchecked")
@@ -726,8 +726,9 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
@Override
public ExecutableSelectStatement select(List<? extends TableNode> from, HintNode hint, boolean isDistinct, List<AliasedNode> select,
ParseNode where, List<ParseNode> groupBy, ParseNode having,
- List<OrderByNode> orderBy, LimitNode limit, int bindCount, boolean isAggregate) {
- return new ExecutableSelectStatement(from, hint, isDistinct, select, where, groupBy == null ? Collections.<ParseNode>emptyList() : groupBy, having, orderBy == null ? Collections.<OrderByNode>emptyList() : orderBy, limit, bindCount, isAggregate);
+ List<OrderByNode> orderBy, LimitNode limit, int bindCount, boolean isAggregate, boolean hasSequence) {
+ return new ExecutableSelectStatement(from, hint, isDistinct, select, where, groupBy == null ? Collections.<ParseNode>emptyList() : groupBy,
+ having, orderBy == null ? Collections.<OrderByNode>emptyList() : orderBy, limit, bindCount, isAggregate, hasSequence);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseContext.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseContext.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseContext.java
index 8ef81c0..b6da901 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseContext.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseContext.java
@@ -23,6 +23,7 @@ import com.google.common.collect.Lists;
public class ParseContext {
private boolean isAggregate;
+ private boolean hasSequences;
public ParseContext() {
}
@@ -35,6 +36,14 @@ public class ParseContext {
this.isAggregate |= isAggregate;
}
+ public boolean hasSequences() {
+ return hasSequences;
+ }
+
+ public void hasSequences(boolean hasSequences) {
+ this.hasSequences |= hasSequences;
+ }
+
public static class Stack {
private final List<ParseContext> stack = Lists.newArrayListWithExpectedSize(5);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
index 088902e..2378f6d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java
@@ -563,9 +563,10 @@ public class ParseNodeFactory {
}
public SelectStatement select(List<? extends TableNode> from, HintNode hint, boolean isDistinct, List<AliasedNode> select, ParseNode where,
- List<ParseNode> groupBy, ParseNode having, List<OrderByNode> orderBy, LimitNode limit, int bindCount, boolean isAggregate) {
+ List<ParseNode> groupBy, ParseNode having, List<OrderByNode> orderBy, LimitNode limit, int bindCount, boolean isAggregate, boolean hasSequence) {
- return new SelectStatement(from, hint, isDistinct, select, where, groupBy == null ? Collections.<ParseNode>emptyList() : groupBy, having, orderBy == null ? Collections.<OrderByNode>emptyList() : orderBy, limit, bindCount, isAggregate);
+ return new SelectStatement(from, hint, isDistinct, select, where, groupBy == null ? Collections.<ParseNode>emptyList() : groupBy, having,
+ orderBy == null ? Collections.<OrderByNode>emptyList() : orderBy, limit, bindCount, isAggregate, hasSequence);
}
public UpsertStatement upsert(NamedTableNode table, HintNode hint, List<ColumnName> columns, List<ParseNode> values, SelectStatement select, int bindCount) {
@@ -577,15 +578,20 @@ public class ParseNodeFactory {
}
public SelectStatement select(SelectStatement statement, ParseNode where, ParseNode having) {
- return select(statement.getFrom(), statement.getHint(), statement.isDistinct(), statement.getSelect(), where, statement.getGroupBy(), having, statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ return select(statement.getFrom(), statement.getHint(), statement.isDistinct(), statement.getSelect(), where, statement.getGroupBy(), having,
+ statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate(), statement.hasSequence());
}
public SelectStatement select(SelectStatement statement, List<? extends TableNode> tables) {
- return select(tables, statement.getHint(), statement.isDistinct(), statement.getSelect(), statement.getWhere(), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ return select(tables, statement.getHint(), statement.isDistinct(), statement.getSelect(), statement.getWhere(), statement.getGroupBy(),
+ statement.getHaving(), statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate(),
+ statement.hasSequence());
}
public SelectStatement select(SelectStatement statement, HintNode hint) {
- return hint == null || hint.isEmpty() ? statement : select(statement.getFrom(), hint, statement.isDistinct(), statement.getSelect(), statement.getWhere(), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ return hint == null || hint.isEmpty() ? statement : select(statement.getFrom(), hint, statement.isDistinct(), statement.getSelect(),
+ statement.getWhere(), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(), statement.getLimit(),
+ statement.getBindCount(), statement.isAggregate(), statement.hasSequence());
}
public SubqueryParseNode subquery(SelectStatement select) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeRewriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeRewriter.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeRewriter.java
index 582ec99..d288de7 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeRewriter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeRewriter.java
@@ -166,7 +166,7 @@ public class ParseNodeRewriter extends TraverseAllParseNodeVisitor<ParseNode> {
}
return NODE_FACTORY.select(normFrom, statement.getHint(), statement.isDistinct(),
normSelectNodes, normWhere, normGroupByNodes, normHaving, normOrderByNodes,
- statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ statement.getLimit(), statement.getBindCount(), statement.isAggregate(), statement.hasSequence());
}
private Map<String, ParseNode> getAliasMap() {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/866d4326/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
index 8270710..6cee588 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/SelectStatement.java
@@ -40,7 +40,7 @@ public class SelectStatement implements FilterableStatement {
Collections.<AliasedNode>singletonList(new AliasedNode(null, LiteralParseNode.ONE)),
null, Collections.<ParseNode>emptyList(),
null, Collections.<OrderByNode>emptyList(),
- null, 0, false);
+ null, 0, false, false);
public static final SelectStatement COUNT_ONE =
new SelectStatement(
Collections.<TableNode>emptyList(), null, false,
@@ -52,14 +52,14 @@ public class SelectStatement implements FilterableStatement {
new BuiltInFunctionInfo(CountAggregateFunction.class, CountAggregateFunction.class.getAnnotation(BuiltInFunction.class))))),
null, Collections.<ParseNode>emptyList(),
null, Collections.<OrderByNode>emptyList(),
- null, 0, true);
+ null, 0, true, false);
public static SelectStatement create(SelectStatement select, HintNode hint) {
if (select.getHint() == hint || hint.isEmpty()) {
return select;
}
return new SelectStatement(select.getFrom(), hint, select.isDistinct(),
select.getSelect(), select.getWhere(), select.getGroupBy(), select.getHaving(),
- select.getOrderBy(), select.getLimit(), select.getBindCount(), select.isAggregate());
+ select.getOrderBy(), select.getLimit(), select.getBindCount(), select.isAggregate(), select.hasSequence());
}
public SelectStatement combine(ParseNode where) {
@@ -71,13 +71,13 @@ public class SelectStatement implements FilterableStatement {
}
return new SelectStatement(this.getFrom(), this.getHint(), this.isDistinct(),
this.getSelect(), where, this.getGroupBy(), this.getHaving(),
- this.getOrderBy(), this.getLimit(), this.getBindCount(), this.isAggregate());
+ this.getOrderBy(), this.getLimit(), this.getBindCount(), this.isAggregate(), this.hasSequence());
}
public static SelectStatement create(SelectStatement select, List<AliasedNode> selects) {
return new SelectStatement(select.getFrom(), select.getHint(), select.isDistinct(),
selects, select.getWhere(), select.getGroupBy(), select.getHaving(),
- select.getOrderBy(), select.getLimit(), select.getBindCount(), select.isAggregate());
+ select.getOrderBy(), select.getLimit(), select.getBindCount(), select.isAggregate(), select.hasSequence());
}
private final List<TableNode> fromTable;
@@ -91,6 +91,7 @@ public class SelectStatement implements FilterableStatement {
private final LimitNode limit;
private final int bindCount;
private final boolean isAggregate;
+ private final boolean hasSequence;
// Count constant expressions
private static int countConstants(List<ParseNode> nodes) {
@@ -103,7 +104,9 @@ public class SelectStatement implements FilterableStatement {
return count;
}
- protected SelectStatement(List<? extends TableNode> from, HintNode hint, boolean isDistinct, List<AliasedNode> select, ParseNode where, List<ParseNode> groupBy, ParseNode having, List<OrderByNode> orderBy, LimitNode limit, int bindCount, boolean isAggregate) {
+ protected SelectStatement(List<? extends TableNode> from, HintNode hint, boolean isDistinct, List<AliasedNode> select,
+ ParseNode where, List<ParseNode> groupBy, ParseNode having, List<OrderByNode> orderBy, LimitNode limit,
+ int bindCount, boolean isAggregate, boolean hasSequence) {
this.fromTable = Collections.unmodifiableList(from);
this.hint = hint == null ? HintNode.EMPTY_HINT_NODE : hint;
this.isDistinct = isDistinct;
@@ -115,6 +118,7 @@ public class SelectStatement implements FilterableStatement {
this.limit = limit;
this.bindCount = bindCount;
this.isAggregate = isAggregate || groupBy.size() != countConstants(groupBy) || this.having != null;
+ this.hasSequence = hasSequence;
}
@Override
@@ -176,6 +180,10 @@ public class SelectStatement implements FilterableStatement {
return isAggregate;
}
+ public boolean hasSequence() {
+ return hasSequence;
+ }
+
@Override
public Operation getOperation() {
return Operation.QUERY;