You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ra...@apache.org on 2014/07/16 06:33:00 UTC
[2/6] Phoenix-933 Local index support to Phoenix (Rajesh)
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java b/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
index 0c69df3..2002137 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
@@ -31,8 +31,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.Type;
import org.apache.hadoop.hbase.client.Delete;
@@ -46,12 +46,15 @@ import org.apache.phoenix.hbase.index.ValueGetter;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
+import org.apache.phoenix.hbase.index.util.IndexManagementUtil.ReferencingColumn;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PColumnFamily;
import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.schema.PDatum;
import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.RowKeySchema;
import org.apache.phoenix.schema.SaltingUtil;
@@ -138,16 +141,28 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
* @param ptr bytes pointer to hold returned serialized value
*/
public static void serialize(PTable dataTable, ImmutableBytesWritable ptr) {
- Iterator<PTable> indexes = nonDisabledIndexIterator(dataTable.getIndexes().iterator());
- if (dataTable.isImmutableRows() || !indexes.hasNext()) {
+ List<PTable> indexes = dataTable.getIndexes();
+ serialize(dataTable, ptr, indexes);
+ }
+
+ /**
+ * For client-side to serialize all IndexMaintainers for a given table
+ * @param dataTable data table
+ * @param ptr bytes pointer to hold returned serialized value
+ * @param indexes indexes to serialize
+ */
+ public static void serialize(PTable dataTable, ImmutableBytesWritable ptr,
+ List<PTable> indexes) {
+ Iterator<PTable> indexesItr = nonDisabledIndexIterator(indexes.iterator());
+ if ((dataTable.isImmutableRows()) || !indexesItr.hasNext()) {
ptr.set(ByteUtil.EMPTY_BYTE_ARRAY);
return;
}
int nIndexes = 0;
int estimatedSize = dataTable.getRowKeySchema().getEstimatedByteSize() + 2;
- while (indexes.hasNext()) {
+ while (indexesItr.hasNext()) {
nIndexes++;
- PTable index = indexes.next();
+ PTable index = indexesItr.next();
estimatedSize += index.getIndexMaintainer(dataTable).getEstimatedByteSize();
}
TrustedByteArrayOutputStream stream = new TrustedByteArrayOutputStream(estimatedSize + 1);
@@ -157,9 +172,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
WritableUtils.writeVInt(output, nIndexes * (dataTable.getBucketNum() == null ? 1 : -1));
// Write out data row key schema once, since it's the same for all index maintainers
dataTable.getRowKeySchema().write(output);
- indexes = nonDisabledIndexIterator(dataTable.getIndexes().iterator());
- while (indexes.hasNext()) {
- indexes.next().getIndexMaintainer(dataTable).write(output);
+ indexesItr = nonDisabledIndexIterator(indexes.iterator());
+ while (indexesItr.hasNext()) {
+ indexesItr.next().getIndexMaintainer(dataTable).write(output);
}
} catch (IOException e) {
throw new RuntimeException(e); // Impossible
@@ -211,6 +226,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
private ImmutableBytesPtr emptyKeyValueCFPtr;
private int nDataCFs;
private boolean indexWALDisabled;
+ private boolean isLocalIndex;
// Transient state
private final boolean isDataTableSalted;
@@ -231,16 +247,29 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this(dataTable.getRowKeySchema(), dataTable.getBucketNum() != null);
this.isMultiTenant = dataTable.isMultiTenant();
this.viewIndexId = index.getViewIndexId() == null ? null : MetaDataUtil.getViewIndexIdDataType().toBytes(index.getViewIndexId());
+ this.isLocalIndex = index.getIndexType() == IndexType.LOCAL;
RowKeySchema dataRowKeySchema = dataTable.getRowKeySchema();
boolean isDataTableSalted = dataTable.getBucketNum() != null;
byte[] indexTableName = index.getPhysicalName().getBytes();
- Integer nIndexSaltBuckets = index.getBucketNum();
+ // Use this for the nDataSaltBuckets as we need this for local indexes
+ // TODO: persist nDataSaltBuckets separately, but maintain b/w compat.
+ Integer nIndexSaltBuckets = isLocalIndex ? dataTable.getBucketNum() : index.getBucketNum();
boolean indexWALDisabled = index.isWALDisabled();
int indexPosOffset = (index.getBucketNum() == null ? 0 : 1) + (this.isMultiTenant ? 1 : 0) + (this.viewIndexId == null ? 0 : 1);
int nIndexColumns = index.getColumns().size() - indexPosOffset;
int nIndexPKColumns = index.getPKColumns().size() - indexPosOffset;
- this.rowKeyMetaData = newRowKeyMetaData(nIndexPKColumns);
+ int indexedColumnsCount = 0;
+ for (int i = indexPosOffset; i<index.getPKColumns().size();i++) {
+ PColumn indexColumn = index.getPKColumns().get(i);
+ PColumn column = IndexUtil.getDataColumn(dataTable, indexColumn.getName().getString());
+ boolean isPKColumn = SchemaUtil.isPKColumn(column);
+ if (!isPKColumn) {
+ indexedColumnsCount++;
+ }
+ }
+ int indexPkColumnCount = this.dataRowKeySchema.getFieldCount() + indexedColumnsCount - (isDataTableSalted ? 1 : 0) - (isMultiTenant ? 1 : 0);
+ this.rowKeyMetaData = newRowKeyMetaData(indexPkColumnCount);
BitSet bitSet = this.rowKeyMetaData.getViewConstantColumnBitSet();
int dataPosOffset = (isDataTableSalted ? 1 : 0) + (this.isMultiTenant ? 1 : 0);
@@ -272,18 +301,32 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this.indexWALDisabled = indexWALDisabled;
}
- public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr) {
+ public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, byte[] regionStartKey, byte[] regionEndKey) {
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- TrustedByteArrayOutputStream stream = new TrustedByteArrayOutputStream(estimatedIndexRowKeyBytes);
+ boolean prependRegionStartKey = isLocalIndex && regionStartKey != null;
+ boolean isIndexSalted = !isLocalIndex && nIndexSaltBuckets > 0;
+ int prefixKeyLength =
+ prependRegionStartKey ? (regionStartKey.length != 0 ? regionStartKey.length
+ : regionEndKey.length) : 0;
+ TrustedByteArrayOutputStream stream = new TrustedByteArrayOutputStream(estimatedIndexRowKeyBytes + (prependRegionStartKey ? prefixKeyLength : 0));
DataOutput output = new DataOutputStream(stream);
try {
- if (nIndexSaltBuckets > 0) {
+ // For local indexes, we must prepend the row key with the start region key
+ if (prependRegionStartKey) {
+ if (regionStartKey.length == 0) {
+ output.write(new byte[prefixKeyLength]);
+ } else {
+ output.write(regionStartKey);
+ }
+ }
+ if (isIndexSalted) {
output.write(0); // will be set at end to index salt byte
}
// The dataRowKeySchema includes the salt byte field,
// so we must adjust for that here.
int dataPosOffset = isDataTableSalted ? 1 : 0 ;
- int nIndexedColumns = getIndexPkColumnCount();
+ BitSet viewConstantColumnBitSet = this.rowKeyMetaData.getViewConstantColumnBitSet();
+ int nIndexedColumns = getIndexPkColumnCount() - getNumViewConstants();
int[][] dataRowKeyLocator = new int[2][nIndexedColumns];
// Skip data table salt byte
int maxRowKeyOffset = rowKeyPtr.getOffset() + rowKeyPtr.getLength();
@@ -300,7 +343,6 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
output.write(viewIndexId);
}
- BitSet viewConstantColumnBitSet = this.rowKeyMetaData.getViewConstantColumnBitSet();
// Write index row key
for (int i = dataPosOffset; i < dataRowKeySchema.getFieldCount(); i++) {
Boolean hasValue=dataRowKeySchema.next(ptr, i, maxRowKeyOffset);
@@ -316,7 +358,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
dataRowKeyLocator[0][pos] = 0;
dataRowKeyLocator[1][pos] = 0;
}
- }
+ }
}
BitSet descIndexColumnBitSet = rowKeyMetaData.getDescIndexColumnBitSet();
int j = 0;
@@ -368,7 +410,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
while (length > minLength && indexRowKey[length-1] == QueryConstants.SEPARATOR_BYTE) {
length--;
}
- if (nIndexSaltBuckets > 0) {
+ if (isIndexSalted) {
// Set salt byte
byte saltByte = SaltingUtil.getSaltingByte(indexRowKey, SaltingUtil.NUM_SALTING_BYTES, length-SaltingUtil.NUM_SALTING_BYTES, nIndexSaltBuckets);
indexRowKey[0] = saltByte;
@@ -385,11 +427,229 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
}
- public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, long ts) throws IOException {
+ /*
+ * Build the data row key from the index row key
+ */
+ public byte[] buildDataRowKey(ImmutableBytesWritable indexRowKeyPtr, byte[][] viewConstants) {
+ RowKeySchema indexRowKeySchema = getIndexRowKeySchema();
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ TrustedByteArrayOutputStream stream = new TrustedByteArrayOutputStream(estimatedIndexRowKeyBytes);
+ DataOutput output = new DataOutputStream(stream);
+ // Increment dataPosOffset until all have been written
+ int dataPosOffset = 0;
+ int viewConstantsIndex = 0;
+ try {
+ int indexPosOffset = !isLocalIndex && nIndexSaltBuckets > 0 ? 1 : 0;
+ int maxRowKeyOffset = indexRowKeyPtr.getOffset() + indexRowKeyPtr.getLength();
+ indexRowKeySchema.iterator(indexRowKeyPtr, ptr, indexPosOffset);
+ if (isDataTableSalted) {
+ dataPosOffset++;
+ output.write(0); // will be set at end to salt byte
+ }
+ if (isMultiTenant) {
+ indexRowKeySchema.next(ptr, indexPosOffset, maxRowKeyOffset);
+ output.write(ptr.get(), ptr.getOffset(), ptr.getLength());
+ if (!dataRowKeySchema.getField(dataPosOffset).getDataType().isFixedWidth()) {
+ output.writeByte(QueryConstants.SEPARATOR_BYTE);
+ }
+ indexPosOffset++;
+ dataPosOffset++;
+ }
+ indexPosOffset = (!isLocalIndex && nIndexSaltBuckets > 0 ? 1 : 0) + (isMultiTenant ? 1 : 0) + (viewIndexId == null ? 0 : 1);
+ BitSet viewConstantColumnBitSet = this.rowKeyMetaData.getViewConstantColumnBitSet();
+ BitSet descIndexColumnBitSet = rowKeyMetaData.getDescIndexColumnBitSet();
+ for (int i = dataPosOffset; i < dataRowKeySchema.getFieldCount(); i++) {
+ // Write view constants from the data table, as these
+ // won't appear in the index (as they're the
+ // same for all rows in this index)
+ if (viewConstantColumnBitSet.get(i)) {
+ output.write(viewConstants[viewConstantsIndex++]);
+ } else {
+ int pos = rowKeyMetaData.getIndexPkPosition(i-dataPosOffset);
+ Boolean hasValue=indexRowKeySchema.iterator(indexRowKeyPtr, ptr, pos + indexPosOffset+1);
+ if (Boolean.TRUE.equals(hasValue)) {
+ // Write data row key value taking into account coercion and inversion
+ // if necessary
+ Field dataField = dataRowKeySchema.getField(i);
+ Field indexField = indexRowKeySchema.getField(pos + indexPosOffset);
+ PDataType indexColumnType = indexField.getDataType();
+ PDataType dataColumnType = dataField.getDataType();
+ SortOrder dataSortOrder = dataField.getSortOrder();
+ SortOrder indexSortOrder = indexField.getSortOrder();
+ boolean isDataColumnInverted = dataSortOrder != SortOrder.ASC;
+ boolean isBytesComparable = dataColumnType.isBytesComparableWith(indexColumnType) ;
+ if (isBytesComparable && isDataColumnInverted == descIndexColumnBitSet.get(pos)) {
+ output.write(ptr.get(), ptr.getOffset(), ptr.getLength());
+ } else {
+ if (!isBytesComparable) {
+ dataColumnType.coerceBytes(ptr, indexColumnType, indexSortOrder, SortOrder.getDefault());
+ }
+ if (descIndexColumnBitSet.get(pos) != isDataColumnInverted) {
+ writeInverted(ptr.get(), ptr.getOffset(), ptr.getLength(), output);
+ } else {
+ output.write(ptr.get(), ptr.getOffset(), ptr.getLength());
+ }
+ }
+ }
+ }
+ if (!dataRowKeySchema.getField(i).getDataType().isFixedWidth() && ((i+1) != dataRowKeySchema.getFieldCount())) {
+ output.writeByte(QueryConstants.SEPARATOR_BYTE);
+ }
+ }
+ int length = stream.size();
+ int minLength = length - maxTrailingNulls;
+ byte[] dataRowKey = stream.getBuffer();
+ // Remove trailing nulls
+ while (length > minLength && dataRowKey[length-1] == QueryConstants.SEPARATOR_BYTE) {
+ length--;
+ }
+ // TODO: need to capture nDataSaltBuckets instead of just a boolean. For now,
+ // we store this in nIndexSaltBuckets, as we only use this function for local indexes
+ // in which case nIndexSaltBuckets would never be used. Note that when we do add this
+ // to be serialized, we have to add it at the end and allow for the value not being
+ // there to maintain compatibility between an old client and a new server.
+ assert(isLocalIndex); // Remove when we persist nDataSaltBuckets
+ if (isDataTableSalted) {
+ // Set salt byte
+ byte saltByte = SaltingUtil.getSaltingByte(dataRowKey, SaltingUtil.NUM_SALTING_BYTES, length-SaltingUtil.NUM_SALTING_BYTES, nIndexSaltBuckets);
+ dataRowKey[0] = saltByte;
+ }
+ return dataRowKey.length == length ? dataRowKey : Arrays.copyOf(dataRowKey, length);
+ } catch (IOException e) {
+ throw new RuntimeException(e); // Impossible
+ } finally {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e); // Impossible
+ }
+ }
+ }
+
+ private volatile RowKeySchema indexRowKeySchema;
+
+ // We have enough information to generate the index row key schema
+ private RowKeySchema generateIndexRowKeySchema() {
+ int nIndexedColumns = getIndexPkColumnCount() + (isMultiTenant ? 1 : 0) + (!isLocalIndex && nIndexSaltBuckets > 0 ? 1 : 0) + (viewIndexId != null ? 1 : 0) - getNumViewConstants();
+ RowKeySchema.RowKeySchemaBuilder builder = new RowKeySchema.RowKeySchemaBuilder(nIndexedColumns);
+ if (!isLocalIndex && nIndexSaltBuckets > 0) {
+ builder.addField(SaltingUtil.SALTING_COLUMN, false, SortOrder.ASC);
+ nIndexedColumns--;
+ }
+ int dataPosOffset = isDataTableSalted ? 1 : 0 ;
+ if (isMultiTenant) {
+ builder.addField(dataRowKeySchema.getField(dataPosOffset++));
+ nIndexedColumns--;
+ }
+ if (viewIndexId != null) {
+ nIndexedColumns--;
+ builder.addField(new PDatum() {
+
+ @Override
+ public boolean isNullable() {
+ return false;
+ }
+
+ @Override
+ public PDataType getDataType() {
+ return MetaDataUtil.getViewIndexIdDataType();
+ }
+
+ @Override
+ public Integer getMaxLength() {
+ return null;
+ }
+
+ @Override
+ public Integer getScale() {
+ return null;
+ }
+
+ @Override
+ public SortOrder getSortOrder() {
+ return SortOrder.getDefault();
+ }
+
+ }, false, SortOrder.getDefault());
+ }
+
+ Field[] indexFields = new Field[nIndexedColumns];
+ BitSet viewConstantColumnBitSet = this.rowKeyMetaData.getViewConstantColumnBitSet();
+ // Add Field for all data row pk columns
+ for (int i = dataPosOffset; i < dataRowKeySchema.getFieldCount(); i++) {
+ // Ignore view constants from the data table, as these
+ // don't need to appear in the index (as they're the
+ // same for all rows in this index)
+ if (!viewConstantColumnBitSet.get(i)) {
+ int pos = rowKeyMetaData.getIndexPkPosition(i-dataPosOffset);
+ indexFields[pos] = dataRowKeySchema.getField(i);
+ }
+ }
+ int indexedColumnTypesIndex = 0;
+ for (Field indexField : indexFields) {
+ if (indexField == null) { // Add field for kv column in index
+ final PDataType dataType = indexedColumnTypes.get(indexedColumnTypesIndex++);
+ builder.addField(new PDatum() {
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ @Override
+ public PDataType getDataType() {
+ return IndexUtil.getIndexColumnDataType(true, dataType);
+ }
+
+ @Override
+ public Integer getMaxLength() {
+ return null;
+ }
+
+ @Override
+ public Integer getScale() {
+ return null;
+ }
+
+ @Override
+ public SortOrder getSortOrder() {
+ return SortOrder.getDefault();
+ }
+
+ }, true, SortOrder.getDefault());
+ } else { // add field from data row key
+ builder.addField(indexField);
+ }
+ }
+ return builder.build();
+ }
+
+ private int getNumViewConstants() {
+ BitSet bitSet = this.rowKeyMetaData.getViewConstantColumnBitSet();
+ int num = 0;
+ for(int i = 0; i<dataRowKeySchema.getFieldCount();i++){
+ if(bitSet.get(i)) num++;
+ }
+ return num;
+ }
+
+ private RowKeySchema getIndexRowKeySchema() {
+ if (indexRowKeySchema != null) {
+ return indexRowKeySchema;
+ }
+ synchronized (this) {
+ if (indexRowKeySchema == null) {
+ indexRowKeySchema = generateIndexRowKeySchema();
+ }
+ }
+ return indexRowKeySchema;
+ }
+
+ public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, long ts, byte[] regionStartKey, byte[] regionEndKey) throws IOException {
Put put = null;
// New row being inserted: add the empty key value
if (valueGetter.getLatestValue(dataEmptyKeyValueRef) == null) {
- byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr);
+ byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey);
put = new Put(indexRowKey);
// add the keyvalue for the empty row
put.add(kvBuilder.buildPut(new ImmutableBytesPtr(indexRowKey),
@@ -401,7 +661,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
for (ColumnReference ref : this.getCoverededColumns()) {
ImmutableBytesPtr cq = this.indexQualifiers.get(i++);
ImmutableBytesPtr value = valueGetter.getLatestValue(ref);
- byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr);
+ byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey);
ImmutableBytesPtr rowKey = new ImmutableBytesPtr(indexRowKey);
if (value != null) {
if (put == null) {
@@ -415,14 +675,6 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
return put;
}
- public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr) throws IOException {
- return buildUpdateMutation(kvBuilder, valueGetter, dataRowKeyPtr, HConstants.LATEST_TIMESTAMP);
- }
-
- public Delete buildDeleteMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, Collection<KeyValue> pendingUpdates) throws IOException {
- return buildDeleteMutation(kvBuilder, valueGetter, dataRowKeyPtr, pendingUpdates, HConstants.LATEST_TIMESTAMP);
- }
-
public boolean isRowDeleted(Collection<KeyValue> pendingUpdates) {
int nDeleteCF = 0;
for (KeyValue kv : pendingUpdates) {
@@ -476,12 +728,12 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
* since we can build the corresponding index row key.
*/
public Delete buildDeleteMutation(KeyValueBuilder kvBuilder, ImmutableBytesWritable dataRowKeyPtr, long ts) throws IOException {
- return buildDeleteMutation(kvBuilder, null, dataRowKeyPtr, Collections.<KeyValue>emptyList(), ts);
+ return buildDeleteMutation(kvBuilder, null, dataRowKeyPtr, Collections.<KeyValue>emptyList(), ts, null, null);
}
@SuppressWarnings("deprecation")
- public Delete buildDeleteMutation(KeyValueBuilder kvBuilder, ValueGetter oldState, ImmutableBytesWritable dataRowKeyPtr, Collection<KeyValue> pendingUpdates, long ts) throws IOException {
- byte[] indexRowKey = this.buildRowKey(oldState, dataRowKeyPtr);
+ public Delete buildDeleteMutation(KeyValueBuilder kvBuilder, ValueGetter oldState, ImmutableBytesWritable dataRowKeyPtr, Collection<KeyValue> pendingUpdates, long ts, byte[] regionStartKey, byte[] regionEndKey) throws IOException {
+ byte[] indexRowKey = this.buildRowKey(oldState, dataRowKeyPtr, regionStartKey, regionEndKey);
// Delete the entire row if any of the indexed columns changed
if (oldState == null || isRowDeleted(pendingUpdates) || hasIndexedColumnChanged(oldState, pendingUpdates)) { // Deleting the entire row
Delete delete = new Delete(indexRowKey, ts);
@@ -521,7 +773,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
return allColumns;
}
- private ImmutableBytesPtr getEmptyKeyValueFamily() {
+ public ImmutableBytesPtr getEmptyKeyValueFamily() {
// Since the metadata of an index table will never change,
// we can infer this based on the family of the first covered column
// If if there are no covered columns, we know it's our default name
@@ -560,7 +812,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
PDataType type = PDataType.values()[WritableUtils.readVInt(input)];
indexedColumnTypes.add(type);
}
- int nCoveredColumns = WritableUtils.readVInt(input);
+ int encodedCoveredolumnsAndLocalIndex = WritableUtils.readVInt(input);
+ isLocalIndex = encodedCoveredolumnsAndLocalIndex < 0;
+ int nCoveredColumns = Math.abs(encodedCoveredolumnsAndLocalIndex) - 1;
coveredColumns = Sets.newLinkedHashSetWithExpectedSize(nCoveredColumns);
for (int i = 0; i < nCoveredColumns; i++) {
byte[] cf = Bytes.readByteArray(input);
@@ -599,7 +853,8 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
PDataType type = indexedColumnTypes.get(i);
WritableUtils.writeVInt(output, type.ordinal());
}
- WritableUtils.writeVInt(output, coveredColumns.size());
+ // Encode coveredColumns.size() and whether or not this is a local index
+ WritableUtils.writeVInt(output, (coveredColumns.size() + 1) * (isLocalIndex ? -1 : 1));
for (ColumnReference ref : coveredColumns) {
Bytes.writeByteArray(output, ref.getFamily());
Bytes.writeByteArray(output, ref.getQualifier());
@@ -643,7 +898,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
private int estimateIndexRowKeyByteSize(int indexColByteSize) {
- int estimatedIndexRowKeyBytes = indexColByteSize + dataRowKeySchema.getEstimatedValueLength() + (nIndexSaltBuckets == 0 || this.isDataTableSalted ? 0 : SaltingUtil.NUM_SALTING_BYTES);
+ int estimatedIndexRowKeyBytes = indexColByteSize + dataRowKeySchema.getEstimatedValueLength() + (nIndexSaltBuckets == 0 || isLocalIndex || this.isDataTableSalted ? 0 : SaltingUtil.NUM_SALTING_BYTES);
return estimatedIndexRowKeyBytes;
}
@@ -669,11 +924,14 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
int nIndexPkColumns = getIndexPkColumnCount();
dataPkPosition = new int[nIndexPkColumns];
Arrays.fill(dataPkPosition, -1);
+ int numViewConstantColumns = 0;
BitSet viewConstantColumnBitSet = rowKeyMetaData.getViewConstantColumnBitSet();
for (int i = dataPkOffset; i < dataRowKeySchema.getFieldCount(); i++) {
if (!viewConstantColumnBitSet.get(i)) {
int dataPkPosition = rowKeyMetaData.getIndexPkPosition(i-dataPkOffset);
this.dataPkPosition[dataPkPosition] = i;
+ } else {
+ numViewConstantColumns++;
}
}
@@ -681,7 +939,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
// We only get rid of nulls for variable length types, so we have to be careful to consider the type of the
// index table, not the data type of the data table
int indexedColumnTypesPos = indexedColumnTypes.size()-1;
- int indexPkPos = nIndexPkColumns-1;
+ int indexPkPos = nIndexPkColumns - numViewConstantColumns - 1;
while (indexPkPos >= 0) {
int dataPkPos = dataPkPosition[indexPkPos];
boolean isDataNullable;
@@ -704,7 +962,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
private int getIndexPkColumnCount() {
- return dataRowKeySchema.getFieldCount() + indexedColumns.size() - (isDataTableSalted ? 1 : 0) - (isMultiTenant ? 1 : 0) - (viewIndexId == null ? 0 : 1);
+ return dataRowKeySchema.getFieldCount() + indexedColumns.size() - (isDataTableSalted ? 1 : 0) - (isMultiTenant ? 1 : 0);
}
private RowKeyMetaData newRowKeyMetaData() {
@@ -859,4 +1117,23 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
public Iterator<ColumnReference> iterator() {
return allColumns.iterator();
}
+
+ public ValueGetter createGetterFromKeyValues(Collection<Cell> pendingUpdates) {
+ final Map<ReferencingColumn, ImmutableBytesPtr> valueMap = Maps.newHashMapWithExpectedSize(pendingUpdates
+ .size());
+ for (Cell kv : pendingUpdates) {
+ // create new pointers to each part of the kv
+ ImmutableBytesPtr family = new ImmutableBytesPtr(kv.getRowArray(),kv.getFamilyOffset(),kv.getFamilyLength());
+ ImmutableBytesPtr qual = new ImmutableBytesPtr(kv.getRowArray(), kv.getQualifierOffset(), kv.getQualifierLength());
+ ImmutableBytesPtr value = new ImmutableBytesPtr(kv.getValueArray(), kv.getValueOffset(), kv.getValueLength());
+ valueMap.put(new ReferencingColumn(family, qual), value);
+ }
+ return new ValueGetter() {
+ @Override
+ public ImmutableBytesPtr getLatestValue(ColumnReference ref) throws IOException {
+ if(ref.equals(dataEmptyKeyValueRef)) return null;
+ return valueMap.get(ReferencingColumn.wrap(ref));
+ }
+ };
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
index 800daf1..f061b8f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
@@ -132,7 +132,7 @@ public class PhoenixIndexCodec extends BaseIndexCodec {
// get the values from the scanner so we can actually use them
ValueGetter valueGetter = IndexManagementUtil.createGetterFromScanner(scanner, dataRowKey);
ptr.set(dataRowKey);
- Put put = maintainer.buildUpdateMutation(kvBuilder, valueGetter, ptr, state.getCurrentTimestamp());
+ Put put = maintainer.buildUpdateMutation(kvBuilder, valueGetter, ptr, state.getCurrentTimestamp(), env.getRegion().getStartKey(), env.getRegion().getEndKey());
indexUpdate.setTable(maintainer.getIndexTableName());
indexUpdate.setUpdate(put);
//make sure we close the scanner when we are done
@@ -162,7 +162,7 @@ public class PhoenixIndexCodec extends BaseIndexCodec {
ptr.set(dataRowKey);
Delete delete =
maintainer.buildDeleteMutation(kvBuilder, valueGetter, ptr,
- state.getPendingUpdate(), state.getCurrentTimestamp());
+ state.getPendingUpdate(), state.getCurrentTimestamp(), env.getRegion().getStartKey(), env.getRegion().getEndKey());
scanner.close();
indexUpdate.setUpdate(delete);
indexUpdates.add(indexUpdate);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelIteratorRegionSplitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelIteratorRegionSplitter.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelIteratorRegionSplitter.java
index 7fb99ff..1498480 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelIteratorRegionSplitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/DefaultParallelIteratorRegionSplitter.java
@@ -41,7 +41,9 @@ import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.query.StatsManager;
import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.util.ReadOnlyProps;
@@ -140,8 +142,7 @@ public class DefaultParallelIteratorRegionSplitter implements ParallelIteratorRe
// distributed across regions, using this scheme compensates for regions that
// have more rows than others, by applying tighter splits and therefore spawning
// off more scans over the overloaded regions.
- int splitsPerRegion = regions.size() >= targetConcurrency ? 1 : (regions.size() > targetConcurrency / 2 ? maxConcurrency : targetConcurrency) / regions.size();
- splitsPerRegion = Math.min(splitsPerRegion, maxIntraRegionParallelization);
+ int splitsPerRegion = getSplitsPerRegion(regions.size());
// Create a multi-map of ServerName to List<KeyRange> which we'll use to round robin from to ensure
// that we keep each region server busy for each query.
ListMultimap<HRegionLocation,KeyRange> keyRangesPerRegion = ArrayListMultimap.create(regions.size(),regions.size() * splitsPerRegion);;
@@ -224,4 +225,13 @@ public class DefaultParallelIteratorRegionSplitter implements ParallelIteratorRe
public List<KeyRange> getSplits() throws SQLException {
return genKeyRanges(getAllRegions());
}
+
+ @Override
+ public int getSplitsPerRegion(int numRegions) {
+ int splitsPerRegion =
+ numRegions >= targetConcurrency ? 1
+ : (numRegions > targetConcurrency / 2 ? maxConcurrency : targetConcurrency)
+ / numRegions;
+ return Math.min(splitsPerRegion, maxIntraRegionParallelization);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java
new file mode 100644
index 0000000..c3a38d5
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java
@@ -0,0 +1,47 @@
+/*
+ * 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.iterate;
+
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.parse.HintNode;
+import org.apache.phoenix.schema.TableRef;
+
+public class LocalIndexParallelIteratorRegionSplitter extends DefaultParallelIteratorRegionSplitter {
+
+ public static DefaultParallelIteratorRegionSplitter getInstance(StatementContext context, TableRef table, HintNode hintNode) {
+ return new LocalIndexParallelIteratorRegionSplitter(context, table, hintNode);
+ }
+
+ protected LocalIndexParallelIteratorRegionSplitter(StatementContext context, TableRef table, HintNode hintNode) {
+ super(context,table,hintNode);
+ }
+
+ @Override
+ protected List<HRegionLocation> getAllRegions() throws SQLException {
+ return context.getConnection().getQueryServices().getAllTableRegions(tableRef.getTable().getPhysicalName().getBytes());
+ }
+
+ @Override
+ public int getSplitsPerRegion(int numRegions) {
+ return 1;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitter.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitter.java
index 1384824..efd9eec 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitter.java
@@ -31,4 +31,7 @@ import org.apache.phoenix.query.KeyRange;
public interface ParallelIteratorRegionSplitter {
public List<KeyRange> getSplits() throws SQLException;
+
+ public int getSplitsPerRegion(int numRegions);
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitterFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitterFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitterFactory.java
index 82f56f1..d3c7d46 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitterFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorRegionSplitterFactory.java
@@ -22,6 +22,7 @@ import java.sql.SQLException;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.parse.HintNode;
import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.PTable.IndexType;
/**
@@ -30,6 +31,9 @@ import org.apache.phoenix.schema.TableRef;
public class ParallelIteratorRegionSplitterFactory {
public static ParallelIteratorRegionSplitter getSplitter(StatementContext context, TableRef table, HintNode hintNode) throws SQLException {
+ if(table.getTable().getIndexType() == IndexType.LOCAL) {
+ return LocalIndexParallelIteratorRegionSplitter.getInstance(context, table, hintNode);
+ }
if (context.getScanRanges().useSkipScanFilter()) {
return SkipRangeParallelIteratorRegionSplitter.getInstance(context, table, hintNode);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
index 34373bd..92fc54a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
@@ -38,6 +38,7 @@ import org.apache.phoenix.parse.*;
import org.apache.phoenix.parse.HintNode.Hint;
import org.apache.phoenix.query.*;
import org.apache.phoenix.schema.*;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTable.ViewType;
import org.apache.phoenix.util.*;
import org.slf4j.Logger;
@@ -223,6 +224,7 @@ public class ParallelIterators extends ExplainTable implements ResultIterators {
List<PeekingResultIterator> iterators = new ArrayList<PeekingResultIterator>(numSplits);
List<Pair<byte[],Future<PeekingResultIterator>>> futures = new ArrayList<Pair<byte[],Future<PeekingResultIterator>>>(numSplits);
final UUID scanId = UUID.randomUUID();
+ final boolean localIndex = this.tableRef.getTable().getType() == PTableType.INDEX && this.tableRef.getTable().getIndexType() == IndexType.LOCAL;
try {
ExecutorService executor = services.getExecutor();
for (KeyRange split : splits) {
@@ -238,7 +240,12 @@ public class ParallelIterators extends ExplainTable implements ResultIterators {
minMaxRange = SaltingUtil.addSaltByte(split.getLowerRange(), minMaxRange);
split = split.intersect(minMaxRange);
}
- }
+ } else if (localIndex) {
+ if (splitScan.getStartRow().length != 0 || splitScan.getStopRow().length != 0) {
+ SaltingUtil.addRegionStartKeyToScanStartAndStopRows(split.getLowerRange(),split.getUpperRange(),
+ splitScan);
+ }
+ }
if (ScanUtil.intersectScanRange(splitScan, split.getLowerRange(), split.getUpperRange(), this.context.getScanRanges().useSkipScanFilter())) {
// Delay the swapping of start/stop row until row so we don't muck with the intersect logic
ScanUtil.swapStartStopRowIfReversed(splitScan);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/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 5f7e017..67b6913 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
@@ -38,6 +38,7 @@ import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.expression.RowKeyColumnExpression;
import org.apache.phoenix.expression.function.ExternalSqlTypeIdFunction;
import org.apache.phoenix.expression.function.IndexStateNameFunction;
+import org.apache.phoenix.expression.function.SQLIndexTypeFunction;
import org.apache.phoenix.expression.function.SQLTableTypeFunction;
import org.apache.phoenix.expression.function.SQLViewTypeFunction;
import org.apache.phoenix.expression.function.SqlTypeNameFunction;
@@ -174,6 +175,8 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
public static final byte[] MULTI_TENANT_BYTES = Bytes.toBytes(MULTI_TENANT);
public static final String VIEW_TYPE = "VIEW_TYPE";
public static final byte[] VIEW_TYPE_BYTES = Bytes.toBytes(VIEW_TYPE);
+ public static final String INDEX_TYPE = "INDEX_TYPE";
+ public static final byte[] INDEX_TYPE_BYTES = Bytes.toBytes(INDEX_TYPE);
public static final String LINK_TYPE = "LINK_TYPE";
public static final byte[] LINK_TYPE_BYTES = Bytes.toBytes(LINK_TYPE);
public static final String ARRAY_SIZE = "ARRAY_SIZE";
@@ -205,7 +208,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
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";
-
+
public static final String TYPE_ID = "TYPE_ID";
private final PhoenixConnection connection;
@@ -855,7 +858,8 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
SALT_BUCKETS + "," +
MULTI_TENANT + "," +
VIEW_STATEMENT + "," +
- SQLViewTypeFunction.NAME + "(" + VIEW_TYPE + ") AS " + VIEW_TYPE +
+ SQLViewTypeFunction.NAME + "(" + VIEW_TYPE + ") AS " + VIEW_TYPE + "," +
+ SQLIndexTypeFunction.NAME + "(" + INDEX_TYPE + ") AS " + INDEX_TYPE +
" from " + SYSTEM_CATALOG + " " + SYSTEM_CATALOG_ALIAS +
" where " + COLUMN_NAME + " is null" +
" and " + COLUMN_FAMILY + " is null");
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/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..eed723a 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
@@ -102,6 +102,7 @@ import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.schema.PDataType;
import org.apache.phoenix.schema.PDatum;
import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.RowKeyValueAccessor;
import org.apache.phoenix.schema.Sequence;
@@ -473,8 +474,8 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
private static class ExecutableCreateIndexStatement extends CreateIndexStatement implements CompilableStatement {
public ExecutableCreateIndexStatement(NamedNode indexName, NamedTableNode dataTable, PrimaryKeyConstraint pkConstraint, List<ColumnName> includeColumns, List<ParseNode> splits,
- ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, int bindCount) {
- super(indexName, dataTable, pkConstraint, includeColumns, splits, props, ifNotExists, bindCount);
+ ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, IndexType indexType, int bindCount) {
+ super(indexName, dataTable, pkConstraint, includeColumns, splits, props, ifNotExists, indexType, bindCount);
}
@SuppressWarnings("unchecked")
@@ -754,8 +755,9 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
}
@Override
- public CreateIndexStatement createIndex(NamedNode indexName, NamedTableNode dataTable, PrimaryKeyConstraint pkConstraint, List<ColumnName> includeColumns, List<ParseNode> splits, ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, int bindCount) {
- return new ExecutableCreateIndexStatement(indexName, dataTable, pkConstraint, includeColumns, splits, props, ifNotExists, bindCount);
+ public CreateIndexStatement createIndex(NamedNode indexName, NamedTableNode dataTable, PrimaryKeyConstraint pkConstraint, List<ColumnName> includeColumns, List<ParseNode> splits,
+ ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, IndexType indexType, int bindCount) {
+ return new ExecutableCreateIndexStatement(indexName, dataTable, pkConstraint, includeColumns, splits, props, ifNotExists, indexType, bindCount);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/join/TupleProjector.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/join/TupleProjector.java b/phoenix-core/src/main/java/org/apache/phoenix/join/TupleProjector.java
index c41125a..41b3906 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/join/TupleProjector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/join/TupleProjector.java
@@ -69,7 +69,7 @@ public class TupleProjector {
valueSet = ValueBitSet.newInstance(schema);
}
- private TupleProjector(KeyValueSchema schema, Expression[] expressions) {
+ public TupleProjector(KeyValueSchema schema, Expression[] expressions) {
this.schema = schema;
this.expressions = expressions;
this.valueSet = ValueBitSet.newInstance(schema);
@@ -203,7 +203,7 @@ public class TupleProjector {
}
public ProjectedValueTuple projectResults(Tuple tuple) {
- byte[] bytesValue = schema.toBytes(tuple, expressions, valueSet, ptr);
+ byte[] bytesValue = schema.toBytes(tuple, getExpressions(), valueSet, ptr);
Cell base = tuple.getValue(0);
return new ProjectedValueTuple(base.getRowArray(), base.getRowOffset(), base.getRowLength(), base.getTimestamp(), bytesValue, valueSet.getEstimatedLength());
}
@@ -238,5 +238,17 @@ public class TupleProjector {
ImmutableBytesWritable keyPtr = dest.getKeyPtr();
return new ProjectedValueTuple(keyPtr.get(), keyPtr.getOffset(), keyPtr.getLength(), dest.getTimestamp(), merged, destBitSetLen);
}
+
+ public KeyValueSchema getSchema() {
+ return schema;
+ }
+
+ public Expression[] getExpressions() {
+ return expressions;
+ }
+
+ public ValueBitSet getValueBitSet() {
+ return valueSet;
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java b/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
index 65c6596..6537485 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
@@ -44,6 +44,7 @@ import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PDatum;
import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableType;
import com.google.common.collect.Lists;
@@ -196,6 +197,15 @@ public class QueryOptimizer {
if (PIndexState.ACTIVE.equals(resolver.getTables().get(0).getTable().getIndexState())) {
QueryCompiler compiler = new QueryCompiler(statement, indexSelect, resolver, targetColumns, parallelIteratorFactory, dataPlan.getContext().getSequenceManager());
QueryPlan plan = compiler.compile();
+ // If query doesn't have where clause and some of columns to project are missing
+ // in the index then we need to get missing columns from main table for each row in
+ // local index. It's like full scan of both local index and data table which is inefficient.
+ // Then we don't use the index. If all the columns to project are present in the index
+ // then we can use the index even the query doesn't have where clause.
+ if (index.getIndexType() == IndexType.LOCAL && indexSelect.getWhere() == null
+ && !plan.getContext().getDataColumns().isEmpty()) {
+ return null;
+ }
// Checking number of columns handles the wildcard cases correctly, as in that case the index
// must contain all columns from the data table to be able to be used.
if (plan.getTableRef().getTable().getIndexState() == PIndexState.ACTIVE && plan.getProjector().getColumnCount() == nColumns) {
@@ -283,27 +293,39 @@ public class QueryOptimizer {
PTable table1 = plan1.getTableRef().getTable();
PTable table2 = plan2.getTableRef().getTable();
int c = plan2.getContext().getScanRanges().getRanges().size() - plan1.getContext().getScanRanges().getRanges().size();
+ boolean bothLocalIndexes = table1.getIndexType() == IndexType.LOCAL && table2.getIndexType() == IndexType.LOCAL;
// Account for potential view constants which are always bound
if (plan1 == dataPlan) { // plan2 is index plan. Ignore the viewIndexId if present
- c += boundRanges - (table2.getViewIndexId() == null ? 0 : 1);
+ c += boundRanges - (table2.getViewIndexId() == null || bothLocalIndexes ? 0 : 1);
+ if(table2.getIndexType()==IndexType.LOCAL && plan2.getContext().getScanRanges().getRanges().size()==0) c++;
} else { // plan1 is index plan. Ignore the viewIndexId if present
- c -= boundRanges - (table1.getViewIndexId() == null ? 0 : 1);
+ c -= boundRanges - (table1.getViewIndexId() == null || bothLocalIndexes ? 0 : 1);
+ if(table1.getIndexType()==IndexType.LOCAL && plan1.getContext().getScanRanges().getRanges().size()==0) c++;
}
if (c != 0) return c;
if (plan1.getGroupBy()!=null && plan2.getGroupBy()!=null) {
if (plan1.getGroupBy().isOrderPreserving() != plan2.getGroupBy().isOrderPreserving()) {
return plan1.getGroupBy().isOrderPreserving() ? -1 : 1;
- }
+ }
}
// Use smaller table (table with fewest kv columns)
c = (table1.getColumns().size() - table1.getPKColumns().size()) - (table2.getColumns().size() - table2.getPKColumns().size());
if (c != 0) return c;
+ // If all things are equal, don't choose local index as it forces scan
+ // on every region.
+ if (table1.getIndexType() == IndexType.LOCAL) {
+ return 1;
+ }
+ if (table2.getIndexType() == IndexType.LOCAL) {
+ return -1;
+ }
+
// All things being equal, just use the table based on the Hint.USE_DATA_OVER_INDEX_TABLE
- if (plan1.getTableRef().getTable().getType() == PTableType.INDEX) {
+ if (table1.getType() == PTableType.INDEX) {
return comparisonOfDataVersusIndexTable;
}
- if (plan2.getTableRef().getTable().getType() == PTableType.INDEX) {
+ if (table2.getType() == PTableType.INDEX) {
return -comparisonOfDataVersusIndexTable;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java
index cc2d971..f688ceb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/parse/CreateIndexStatement.java
@@ -21,6 +21,7 @@ import java.util.Collections;
import java.util.List;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.schema.PTable.IndexType;
import com.google.common.collect.ListMultimap;
@@ -32,10 +33,11 @@ public class CreateIndexStatement extends SingleTableStatement {
private final List<ParseNode> splitNodes;
private final ListMultimap<String,Pair<String,Object>> props;
private final boolean ifNotExists;
+ private final IndexType indexType;
public CreateIndexStatement(NamedNode indexTableName, NamedTableNode dataTable,
PrimaryKeyConstraint indexConstraint, List<ColumnName> includeColumns, List<ParseNode> splits,
- ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, int bindCount) {
+ ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, IndexType indexType, int bindCount) {
super(dataTable, bindCount);
this.indexTableName =TableName.create(dataTable.getName().getSchemaName(),indexTableName.getName());
this.indexConstraint = indexConstraint == null ? PrimaryKeyConstraint.EMPTY : indexConstraint;
@@ -43,6 +45,7 @@ public class CreateIndexStatement extends SingleTableStatement {
this.splitNodes = splits == null ? Collections.<ParseNode>emptyList() : splits;
this.props = props;
this.ifNotExists = ifNotExists;
+ this.indexType = indexType;
}
public PrimaryKeyConstraint getIndexConstraint() {
@@ -69,4 +72,9 @@ public class CreateIndexStatement extends SingleTableStatement {
return ifNotExists;
}
+
+ public IndexType getIndexType() {
+ return indexType;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/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..a17c761 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
@@ -41,6 +41,7 @@ import org.apache.phoenix.parse.FunctionParseNode.BuiltInFunctionInfo;
import org.apache.phoenix.parse.JoinTableNode.JoinType;
import org.apache.phoenix.schema.PDataType;
import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TypeMismatchException;
@@ -268,8 +269,8 @@ public class ParseNodeFactory {
return new CreateTableStatement(tableName, props, columns, pkConstraint, splits, tableType, ifNotExists, baseTableName, tableTypeIdNode, bindCount);
}
- public CreateIndexStatement createIndex(NamedNode indexName, NamedTableNode dataTable, PrimaryKeyConstraint pkConstraint, List<ColumnName> includeColumns, List<ParseNode> splits, ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, int bindCount) {
- return new CreateIndexStatement(indexName, dataTable, pkConstraint, includeColumns, splits, props, ifNotExists, bindCount);
+ public CreateIndexStatement createIndex(NamedNode indexName, NamedTableNode dataTable, PrimaryKeyConstraint pkConstraint, List<ColumnName> includeColumns, List<ParseNode> splits, ListMultimap<String,Pair<String,Object>> props, boolean ifNotExists, IndexType indexType, int bindCount) {
+ return new CreateIndexStatement(indexName, dataTable, pkConstraint, includeColumns, splits, props, ifNotExists, indexType, bindCount);
}
public CreateSequenceStatement createSequence(TableName tableName, ParseNode startsWith, ParseNode incrementBy, ParseNode cacheSize, boolean ifNotExits, int bindCount){
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/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..db0936d 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
@@ -436,7 +436,12 @@ public class ParseNodeRewriter extends TraverseAllParseNodeVisitor<ParseNode> {
@Override
public ParseNode visitLeave(StringConcatParseNode node, List<ParseNode> l) throws SQLException {
- return node;
+ return leaveCompoundNode(node, l, new CompoundNodeFactory() {
+ @Override
+ public ParseNode createNode(List<ParseNode> children) {
+ return NODE_FACTORY.concat(children);
+ }
+ });
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/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 aa6fe17..0d9553f 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
@@ -123,6 +123,7 @@ import org.apache.phoenix.exception.PhoenixIOException;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.hbase.index.IndexRegionSplitPolicy;
import org.apache.phoenix.hbase.index.Indexer;
import org.apache.phoenix.hbase.index.covered.CoveredColumnsIndexBuilder;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
@@ -803,6 +804,9 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (isMetaTable) {
newDesc.remove(HTableDescriptor.SPLIT_POLICY);
}
+ if (newDesc.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_BYTES) != null && Boolean.TRUE.equals(PDataType.BOOLEAN.toObject(newDesc.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_BYTES)))) {
+ newDesc.setValue(HTableDescriptor.SPLIT_POLICY, IndexRegionSplitPolicy.class.getName());
+ }
try {
if (splits == null) {
admin.createTable(newDesc);
@@ -1019,6 +1023,40 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
}
}
+ private void ensureLocalIndexTableCreated(byte[] physicalTableName, Map<String,Object> tableProps, List<Pair<byte[],Map<String,Object>>> families, byte[][] splits, long timestamp) throws SQLException {
+ PTable table;
+ String parentTableName = Bytes.toString(physicalTableName, MetaDataUtil.LOCAL_INDEX_TABLE_PREFIX_BYTES.length,
+ physicalTableName.length - MetaDataUtil.LOCAL_INDEX_TABLE_PREFIX_BYTES.length);
+ try {
+ table = latestMetaData.getTable(new PTableKey(PName.EMPTY_NAME, parentTableName));
+ if (table.getTimeStamp() >= timestamp) { // Table in cache is newer than client timestamp which shouldn't be the case
+ throw new TableNotFoundException(table.getSchemaName().getString(), table.getTableName().getString());
+ }
+ } catch (TableNotFoundException e) {
+ byte[] schemaName = Bytes.toBytes(SchemaUtil.getSchemaNameFromFullName(parentTableName));
+ byte[] tableName = Bytes.toBytes(SchemaUtil.getTableNameFromFullName(parentTableName));
+ MetaDataMutationResult result = this.getTable(null, schemaName, tableName, HConstants.LATEST_TIMESTAMP, timestamp);
+ table = result.getTable();
+ if (table == null) {
+ throw e;
+ }
+ }
+ ensureLocalIndexTableCreated(physicalTableName, tableProps, families, splits);
+ }
+
+ private void ensureLocalIndexTableCreated(byte[] physicalTableName, Map<String, Object> tableProps, List<Pair<byte[], Map<String, Object>>> families, byte[][] splits) throws SQLException, TableAlreadyExistsException {
+ tableProps.put(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_NAME, TRUE_BYTES_AS_STRING);
+ HTableDescriptor desc = ensureTableCreated(physicalTableName, PTableType.TABLE, tableProps, families, splits, true);
+ if (desc != null) {
+ if (!Boolean.TRUE.equals(PDataType.BOOLEAN.toObject(desc.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_BYTES)))) {
+ String fullTableName = Bytes.toString(physicalTableName);
+ throw new TableAlreadyExistsException(
+ "Unable to create shared physical table for local indexes.",
+ SchemaUtil.getSchemaNameFromFullName(fullTableName),
+ SchemaUtil.getTableNameFromFullName(fullTableName));
+ }
+ }
+ }
private boolean ensureViewIndexTableDropped(byte[] physicalTableName, long timestamp) throws SQLException {
byte[] physicalIndexName = MetaDataUtil.getViewIndexPhysicalName(physicalTableName);
@@ -1052,7 +1090,40 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
}
return wasDeleted;
}
-
+
+ private boolean ensureLocalIndexTableDropped(byte[] physicalTableName, long timestamp) throws SQLException {
+ byte[] physicalIndexName = MetaDataUtil.getLocalIndexPhysicalName(physicalTableName);
+ HTableDescriptor desc = null;
+ HBaseAdmin admin = null;
+ boolean wasDeleted = false;
+ try {
+ admin = new HBaseAdmin(config);
+ try {
+ desc = admin.getTableDescriptor(physicalIndexName);
+ if (Boolean.TRUE.equals(PDataType.BOOLEAN.toObject(desc.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_BYTES)))) {
+ final ReadOnlyProps props = this.getProps();
+ final boolean dropMetadata = props.getBoolean(DROP_METADATA_ATTRIB, DEFAULT_DROP_METADATA);
+ if (dropMetadata) {
+ admin.disableTable(physicalIndexName);
+ admin.deleteTable(physicalIndexName);
+ wasDeleted = true;
+ }
+ }
+ } catch (org.apache.hadoop.hbase.TableNotFoundException ignore) {
+ // Ignore, as we may never have created a view index table
+ }
+ } catch (IOException e) {
+ throw ServerUtil.parseServerException(e);
+ } finally {
+ try {
+ if (admin != null) admin.close();
+ } catch (IOException e) {
+ logger.warn("",e);
+ }
+ }
+ return wasDeleted;
+ }
+
@Override
public MetaDataMutationResult createTable(final List<Mutation> tableMetaData, byte[] physicalTableName, PTableType tableType,
Map<String,Object> tableProps, final List<Pair<byte[],Map<String,Object>>> families, byte[][] splits) throws SQLException {
@@ -1064,16 +1135,23 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
byte[] schemaBytes = rowKeyMetadata[PhoenixDatabaseMetaData.SCHEMA_NAME_INDEX];
byte[] tableBytes = rowKeyMetadata[PhoenixDatabaseMetaData.TABLE_NAME_INDEX];
byte[] tableName = physicalTableName != null ? physicalTableName : SchemaUtil.getTableNameAsBytes(schemaBytes, tableBytes);
+ boolean localIndexTable = Boolean.TRUE.equals(tableProps.remove(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_NAME));
+
if ((tableType == PTableType.VIEW && physicalTableName != null) || (tableType != PTableType.VIEW && physicalTableName == null)) {
// For views this will ensure that metadata already exists
+ // For tables and indexes, this will create the metadata if it doesn't already exist
ensureTableCreated(tableName, tableType, tableProps, families, splits, true);
}
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- if (tableType == PTableType.INDEX && physicalTableName != null) { // Index on view
+ if (tableType == PTableType.INDEX) { // Index on view
// Physical index table created up front for multi tenant
// TODO: if viewIndexId is Short.MIN_VALUE, then we don't need to attempt to create it
- if (!MetaDataUtil.isMultiTenant(m, kvBuilder, ptr)) {
- ensureViewIndexTableCreated(tenantIdBytes.length == 0 ? null : PNameFactory.newName(tenantIdBytes), physicalTableName, MetaDataUtil.getClientTimeStamp(m));
+ if (physicalTableName != null && !MetaDataUtil.isMultiTenant(m, kvBuilder, ptr)) {
+ if (localIndexTable) {
+ ensureLocalIndexTableCreated(tableName, tableProps, families, splits, MetaDataUtil.getClientTimeStamp(m));
+ } else {
+ ensureViewIndexTableCreated(tenantIdBytes.length == 0 ? null : PNameFactory.newName(tenantIdBytes), physicalTableName, MetaDataUtil.getClientTimeStamp(m));
+ }
}
} else if (tableType == PTableType.TABLE && MetaDataUtil.isMultiTenant(m, kvBuilder, ptr)) { // Create view index table up front for multi tenant tables
ptr.set(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES);
@@ -1095,6 +1173,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
familiesPlusDefault.add(new Pair<byte[],Map<String,Object>>(defaultCF,Collections.<String,Object>emptyMap()));
}
ensureViewIndexTableCreated(tableName, tableProps, familiesPlusDefault, MetaDataUtil.isSalted(m, kvBuilder, ptr) ? splits : null, MetaDataUtil.getClientTimeStamp(m));
+ ensureLocalIndexTableCreated(MetaDataUtil.getLocalIndexPhysicalName(tableName), tableProps, familiesPlusDefault, splits);
}
byte[] tableKey = SchemaUtil.getTableKey(tenantIdBytes, schemaBytes, tableBytes);
@@ -1190,6 +1269,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
byte[] physicalTableName = SchemaUtil.getTableNameAsBytes(schemaBytes, tableBytes);
long timestamp = MetaDataUtil.getClientTimeStamp(tableMetaData);
ensureViewIndexTableDropped(physicalTableName, timestamp);
+ ensureLocalIndexTableDropped(physicalTableName, timestamp);
}
break;
default:
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/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 db411ab..bf80630 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
@@ -36,6 +36,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DISABLE_WAL;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_ROWS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INCREMENT_BY;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_STATE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_AUTOINCREMENT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_NULLABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_VIEW_REFERENCED;
@@ -207,6 +208,7 @@ public interface QueryConstants {
SCOPE_TABLE + " VARCHAR," +
SOURCE_DATA_TYPE + " SMALLINT," +
IS_AUTOINCREMENT + " VARCHAR," +
+ INDEX_TYPE + " UNSIGNED_TINYINT," +
"CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + ","
+ TABLE_SCHEM + "," + TABLE_NAME + "," + COLUMN_NAME + "," + COLUMN_FAMILY + "))\n" +
HConstants.VERSIONS + "=" + MetaDataProtocol.DEFAULT_MAX_META_DATA_VERSIONS + ",\n" +
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
index 0910712..7a00082 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
@@ -36,17 +36,21 @@ import org.apache.phoenix.util.SchemaUtil;
* @since 0.1
*/
@Immutable
-public final class ColumnRef {
+public class ColumnRef {
private final TableRef tableRef;
private final int columnPosition;
private final int pkSlotPosition;
- public ColumnRef(ColumnRef columnRef, long timeStamp) {
+ protected ColumnRef(ColumnRef columnRef, long timeStamp) {
this.tableRef = new TableRef(columnRef.tableRef, timeStamp);
this.columnPosition = columnRef.columnPosition;
this.pkSlotPosition = columnRef.pkSlotPosition;
}
+ public ColumnRef(TableRef tableRef, String familyName, String columnName) throws MetaDataEntityNotFoundException {
+ this(tableRef, tableRef.getTable().getColumnFamily(familyName).getColumn(columnName).getPosition());
+ }
+
public ColumnRef(TableRef tableRef, int columnPosition) {
if (tableRef == null) {
throw new NullPointerException();
@@ -113,7 +117,12 @@ public final class ColumnRef {
String displayName = SchemaUtil.getColumnDisplayName(defaultFamilyName.equals(dataFamilyName) ? null : dataFamilyName, dataColumnName);
return new KeyValueColumnExpression(column, displayName);
}
-
+
+ // TODO: In ExpressionCompiler create a ColumnRef for a local index that causes a
+ // different kind of ColumnExpression to be created here. You might be able to
+ // use ProjectedColumnExpression, but not sure. The column values from the data
+ // table should get returned in a single KeyValue in a similar format (using a
+ // KeyValueSchema).
if (table.getType() == PTableType.JOIN) {
return new ProjectedColumnExpression(column, table, column.getName().getString());
}
@@ -123,6 +132,10 @@ public final class ColumnRef {
return new KeyValueColumnExpression(column, displayName);
}
+ public ColumnRef cloneAtTimestamp(long timestamp) {
+ return new ColumnRef(this, timestamp);
+ }
+
public int getColumnPosition() {
return columnPosition;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/schema/LocalIndexDataColumnRef.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/LocalIndexDataColumnRef.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/LocalIndexDataColumnRef.java
new file mode 100644
index 0000000..16fe1a5
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/LocalIndexDataColumnRef.java
@@ -0,0 +1,55 @@
+package org.apache.phoenix.schema;
+
+import java.sql.SQLException;
+import java.util.Set;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.compile.FromCompiler;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.expression.ColumnExpression;
+import org.apache.phoenix.expression.ProjectedColumnExpression;
+import org.apache.phoenix.parse.ParseNodeFactory;
+import org.apache.phoenix.parse.TableName;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.SchemaUtil;
+
+public class LocalIndexDataColumnRef extends ColumnRef {
+ final private int position;
+ final private Set<PColumn> columns;
+ private static final ParseNodeFactory FACTORY = new ParseNodeFactory();
+
+ public LocalIndexDataColumnRef(StatementContext context, String indexColumnName) throws MetaDataEntityNotFoundException, SQLException {
+ super(FromCompiler.getResolver(
+ FACTORY.namedTable(null, TableName.create(context.getCurrentTable().getTable()
+ .getSchemaName().getString(), context.getCurrentTable().getTable()
+ .getParentTableName().getString())), context.getConnection()).resolveTable(
+ context.getCurrentTable().getTable().getSchemaName().getString(),
+ context.getCurrentTable().getTable().getParentTableName().getString()), IndexUtil
+ .getDataColumnFamilyName(indexColumnName), IndexUtil
+ .getDataColumnName(indexColumnName));
+ position = context.getDataColumnPosition(this.getColumn());
+ columns = context.getDataColumns();
+ }
+
+ protected LocalIndexDataColumnRef(LocalIndexDataColumnRef localIndexDataColumnRef, long timestamp) {
+ super(localIndexDataColumnRef, timestamp);
+ this.position = localIndexDataColumnRef.position;
+ this.columns = localIndexDataColumnRef.columns;
+ }
+
+ @Override
+ public ColumnRef cloneAtTimestamp(long timestamp) {
+ return new LocalIndexDataColumnRef(this, timestamp);
+ }
+
+ @Override
+ public ColumnExpression newColumnExpression() {
+ PTable table = this.getTable();
+ PColumn column = this.getColumn();
+ // TODO: util for this or store in member variable
+ byte[] defaultFamily = table.getDefaultFamilyName() == null ? QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES : table.getDefaultFamilyName().getBytes();
+ String displayName = SchemaUtil.getColumnDisplayName(Bytes.compareTo(defaultFamily, column.getFamilyName().getBytes()) == 0 ? null : column.getFamilyName().getBytes(), column.getName().getBytes());
+ return new ProjectedColumnExpression(this.getColumn(), columns, position, displayName);
+ }
+}