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/04/28 18:31:19 UTC
git commit: PHOENIX-955 Skip region start key at beginning of local
index rows (JamesTaylor)
Repository: incubator-phoenix
Updated Branches:
refs/heads/local-index ef50aab11 -> 360f5a1cd
PHOENIX-955 Skip region start key at beginning of local index rows (JamesTaylor)
Project: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/commit/360f5a1c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/tree/360f5a1c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-phoenix/diff/360f5a1c
Branch: refs/heads/local-index
Commit: 360f5a1cd6ca7baeba84d4fab09b5b496113ab78
Parents: ef50aab
Author: James Taylor <ja...@apache.org>
Authored: Mon Apr 28 09:31:31 2014 -0700
Committer: James Taylor <ja...@apache.org>
Committed: Mon Apr 28 09:31:31 2014 -0700
----------------------------------------------------------------------
.../index/balancer/TestIndexLoadBalancer.java | 4 --
.../coprocessor/BaseScannerRegionObserver.java | 1 +
.../GroupedAggregateRegionObserver.java | 20 ++++++++--
.../phoenix/coprocessor/ScanRegionObserver.java | 19 ++++++++--
.../apache/phoenix/execute/BasicQueryPlan.java | 6 +++
.../expression/RowKeyColumnExpression.java | 12 +++++-
.../visitor/RowKeyExpressionVisitor.java | 36 ++++++++++++++++++
.../apache/phoenix/filter/SkipScanFilter.java | 7 +++-
.../apache/phoenix/index/IndexMaintainer.java | 40 ++++++++++----------
.../apache/phoenix/index/PhoenixIndexCodec.java | 4 +-
.../java/org/apache/phoenix/util/IndexUtil.java | 24 +++++++++++-
.../java/org/apache/phoenix/util/ScanUtil.java | 35 +++++++++++++++++
.../phoenix/index/IndexMaintainerTest.java | 2 +-
13 files changed, 174 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/it/java/org/apache/phoenix/hbase/index/balancer/TestIndexLoadBalancer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/hbase/index/balancer/TestIndexLoadBalancer.java b/phoenix-core/src/it/java/org/apache/phoenix/hbase/index/balancer/TestIndexLoadBalancer.java
index 7a5d61c..9020e73 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/hbase/index/balancer/TestIndexLoadBalancer.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/hbase/index/balancer/TestIndexLoadBalancer.java
@@ -23,8 +23,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HColumnDescriptor;
@@ -62,8 +60,6 @@ import org.junit.experimental.categories.Category;
@Category(LargeTests.class)
public class TestIndexLoadBalancer {
- private static final Log LOG = LogFactory.getLog(TestIndexLoadBalancer.class);
-
private static HBaseTestingUtility UTIL = new HBaseTestingUtility();
private static HBaseAdmin admin = null;
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index b94bf0a..6f0124a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -44,6 +44,7 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
public static final String EMPTY_CF = "_EmptyCF";
public static final String SPECIFIC_ARRAY_INDEX = "_SpecificArrayIndex";
public static final String GROUP_BY_LIMIT = "_GroupByLimit";
+ public static final String LOCAL_INDEX = "_LocalIndex";
/**
* Used by logger to identify coprocessor
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
index ae3c1c4..d2aaf84 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/GroupedAggregateRegionObserver.java
@@ -46,7 +46,6 @@ import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.regionserver.HRegion;
-import org.apache.hadoop.hbase.regionserver.MultiVersionConsistencyControl;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.WritableUtils;
@@ -60,12 +59,12 @@ import org.apache.phoenix.expression.aggregator.ServerAggregators;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.join.HashJoinInfo;
import org.apache.phoenix.join.ScanProjector;
-import org.apache.phoenix.memory.GlobalMemoryManager;
import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.PDataType;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
+import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.SizedUtil;
@@ -107,7 +106,17 @@ public class GroupedAggregateRegionObserver extends BaseScannerRegionObserver {
}
keyOrdered = true;
}
- List<Expression> expressions = deserializeGroupByExpressions(expressionBytes);
+ int offset = 0;
+ if (ScanUtil.isLocalIndex(scan)) {
+ /*
+ * For local indexes, we need to set an offset on row key expressions to skip
+ * the region start key.
+ */
+ offset = c.getEnvironment().getRegion().getStartKey().length;
+ ScanUtil.setRowKeyOffset(scan, offset);
+ }
+
+ List<Expression> expressions = deserializeGroupByExpressions(expressionBytes, offset);
ServerAggregators aggregators =
ServerAggregators.deserialize(scan
.getAttribute(BaseScannerRegionObserver.AGGREGATORS), c
@@ -167,7 +176,7 @@ public class GroupedAggregateRegionObserver extends BaseScannerRegionObserver {
}
- private List<Expression> deserializeGroupByExpressions(byte[] expressionBytes)
+ private List<Expression> deserializeGroupByExpressions(byte[] expressionBytes, int offset)
throws IOException {
List<Expression> expressions = new ArrayList<Expression>(3);
ByteArrayInputStream stream = new ByteArrayInputStream(expressionBytes);
@@ -179,6 +188,9 @@ public class GroupedAggregateRegionObserver extends BaseScannerRegionObserver {
Expression expression =
ExpressionType.values()[expressionOrdinal].newInstance();
expression.readFields(input);
+ if (offset != 0) {
+ IndexUtil.setRowKeyExpressionOffset(expression, offset);
+ }
expressions.add(expression);
} catch (EOFException e) {
break;
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
index 84d6b0e..8c3c470 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
@@ -25,8 +25,8 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.Type;
@@ -57,6 +57,7 @@ import org.apache.phoenix.schema.PDataType;
import org.apache.phoenix.schema.ValueBitSet;
import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.ServerUtil;
@@ -100,7 +101,7 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
}
}
- public static OrderedResultIterator deserializeFromScan(Scan scan, RegionScanner s) {
+ public static OrderedResultIterator deserializeFromScan(Scan scan, RegionScanner s, int offset) {
byte[] topN = scan.getAttribute(BaseScannerRegionObserver.TOPN);
if (topN == null) {
return null;
@@ -116,6 +117,9 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
for (int i = 0; i < size; i++) {
OrderByExpression orderByExpression = new OrderByExpression();
orderByExpression.readFields(input);
+ if (offset != 0) {
+ IndexUtil.setRowKeyExpressionOffset(orderByExpression.getExpression(), offset);
+ }
orderByExpressions.add(orderByExpression);
}
ResultIterator inner = new RegionScannerResultIterator(s);
@@ -176,6 +180,15 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
if (isScanQuery == null || Bytes.compareTo(PDataType.FALSE_BYTES, isScanQuery) == 0) {
return s;
}
+ int offset = 0;
+ if (ScanUtil.isLocalIndex(scan)) {
+ /*
+ * For local indexes, we need to set an offset on row key expressions to skip
+ * the region start key.
+ */
+ offset = c.getEnvironment().getRegion().getStartKey().length;
+ ScanUtil.setRowKeyOffset(scan, offset);
+ }
final ScanProjector p = ScanProjector.deserializeProjectorFromScan(scan);
final HashJoinInfo j = HashJoinInfo.deserializeHashJoinFromScan(scan);
@@ -186,7 +199,7 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
innerScanner = new HashJoinRegionScanner(s, p, j, tenantId, c.getEnvironment());
}
- final OrderedResultIterator iterator = deserializeFromScan(scan,innerScanner);
+ final OrderedResultIterator iterator = deserializeFromScan(scan,innerScanner, offset);
List<KeyValueColumnExpression> arrayKVRefs = new ArrayList<KeyValueColumnExpression>();
Expression[] arrayFuncRefs = deserializeArrayPostionalExpressionInfoFromScan(
scan, innerScanner, arrayKVRefs);
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
index cc1193c..f77eb8d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
@@ -35,6 +35,7 @@ import org.apache.phoenix.iterate.ParallelIterators.ParallelIteratorFactory;
import org.apache.phoenix.iterate.ResultIterator;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.util.SQLCloseable;
import org.apache.phoenix.util.SQLCloseables;
@@ -130,6 +131,8 @@ public abstract class BasicQueryPlan implements QueryPlan {
return ResultIterator.EMPTY_ITERATOR;
}
+ // Set miscellaneous scan attributes. This is the last chance to set them before we
+ // clone the scan for each parallelized chunk.
Scan scan = context.getScan();
// Set producer on scan so HBase server does round robin processing
//setProducer(scan);
@@ -141,6 +144,9 @@ public abstract class BasicQueryPlan implements QueryPlan {
Long scn = connection.getSCN();
ScanUtil.setTimeRange(scan, scn == null ? context.getCurrentTime() : scn);
ScanUtil.setTenantId(scan, connection.getTenantId() == null ? null : connection.getTenantId().getBytes());
+ if (context.getCurrentTable().getTable().getIndexType() == IndexType.LOCAL) {
+ ScanUtil.setLocalIndex(scan);
+ }
ResultIterator iterator = newIterator();
return dependencies.isEmpty() ?
iterator : new DelegateResultIterator(iterator) {
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
index 6761733..7913ab8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
@@ -41,6 +41,7 @@ public class RowKeyColumnExpression extends ColumnExpression {
private PDataType fromType;
private RowKeyValueAccessor accessor;
protected final String name;
+ private int offset;
public RowKeyColumnExpression() {
name = null; // Only on client
@@ -65,6 +66,15 @@ public class RowKeyColumnExpression extends ColumnExpression {
this(datum, accessor, fromType, datum.toString());
}
+ /**
+ * Used to set an offset to be skipped from the start of a the row key. Used by
+ * local indexing to skip the region start key bytes.
+ * @param offset the number of bytes to offset accesses to row key columns
+ */
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
+
public int getPosition() {
return accessor.getIndex();
}
@@ -94,7 +104,7 @@ public class RowKeyColumnExpression extends ColumnExpression {
@Override
public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) {
tuple.getKey(ptr);
- int offset = accessor.getOffset(ptr.get(), ptr.getOffset());
+ int offset = accessor.getOffset(ptr.get(), ptr.getOffset() + this.offset);
// Null is represented in the last expression of a multi-part key
// by the bytes not being present.
int maxOffset = ptr.getOffset() + ptr.getLength();
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java
new file mode 100644
index 0000000..a3e8fb0
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java
@@ -0,0 +1,36 @@
+/*
+ * 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.expression.visitor;
+
+import org.apache.phoenix.expression.RowKeyColumnExpression;
+
+
+
+
+/**
+ *
+ * Implementation of ExpressionVisitor where only a RowKeyColumnExpression (i.e.
+ * a reference to a column that makes up the row key) is being visited,
+ *
+ *
+ * @since 0.1
+ */
+public abstract class RowKeyExpressionVisitor extends TraverseAllExpressionVisitor<Void> {
+ @Override
+ abstract public Void visit(RowKeyColumnExpression node);
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java b/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
index 2551e83..8f6a337 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
@@ -72,6 +72,7 @@ public class SkipScanFilter extends FilterBase implements Writable {
private byte[] endKey;
private int endKeyLength;
private boolean isDone;
+ private int offset;
private final ImmutableBytesWritable ptr = new ImmutableBytesWritable();
@@ -85,6 +86,10 @@ public class SkipScanFilter extends FilterBase implements Writable {
public SkipScanFilter(List<List<KeyRange>> slots, RowKeySchema schema) {
init(slots, schema);
}
+
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
private void init(List<List<KeyRange>> slots, RowKeySchema schema) {
for (List<KeyRange> ranges : slots) {
@@ -113,7 +118,7 @@ public class SkipScanFilter extends FilterBase implements Writable {
@Override
public ReturnCode filterKeyValue(Cell kv) {
- return navigate(kv.getRowArray(), kv.getRowOffset(),kv.getRowLength(),Terminate.AFTER);
+ return navigate(kv.getRowArray(), kv.getRowOffset() + offset,kv.getRowLength(),Terminate.AFTER);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/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 c1d6c25..1ca330f 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
@@ -32,7 +32,6 @@ import java.util.Map;
import java.util.Set;
import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
@@ -51,6 +50,7 @@ import org.apache.phoenix.schema.PColumnFamily;
import org.apache.phoenix.schema.PDataType;
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;
@@ -210,6 +210,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;
@@ -269,13 +270,19 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
this.emptyKeyValueCFPtr = SchemaUtil.getEmptyColumnFamilyPtr(index);
this.nDataCFs = dataTable.getColumnFamilies().size();
this.indexWALDisabled = indexWALDisabled;
+ this.isLocalIndex = index.getIndexType() == IndexType.LOCAL;
}
- public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr) {
+ public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, byte[] regionStartKey) {
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- TrustedByteArrayOutputStream stream = new TrustedByteArrayOutputStream(estimatedIndexRowKeyBytes);
+ boolean prependRegionStartKey = isLocalIndex && regionStartKey != null;
+ TrustedByteArrayOutputStream stream = new TrustedByteArrayOutputStream(estimatedIndexRowKeyBytes + (prependRegionStartKey ? regionStartKey.length : 0));
DataOutput output = new DataOutputStream(stream);
try {
+ // For local indexes, we must prepend the row key with the start region key
+ if (prependRegionStartKey) {
+ output.write(regionStartKey);
+ }
if (nIndexSaltBuckets > 0) {
output.write(0); // will be set at end to index salt byte
}
@@ -384,11 +391,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
}
- public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, long ts) throws IOException {
+ public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, long ts, byte[] regionStartKey) 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);
put = new Put(indexRowKey);
// add the keyvalue for the empty row
put.add(kvBuilder.buildPut(new ImmutableBytesPtr(indexRowKey),
@@ -400,7 +407,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);
ImmutableBytesPtr rowKey = new ImmutableBytesPtr(indexRowKey);
if (value != null) {
if (put == null) {
@@ -414,14 +421,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) {
@@ -467,12 +466,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);
}
@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) throws IOException {
+ byte[] indexRowKey = this.buildRowKey(oldState, dataRowKeyPtr, regionStartKey);
// 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);
@@ -551,7 +550,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);
@@ -590,7 +591,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());
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/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..119c7cf 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());
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());
scanner.close();
indexUpdate.setUpdate(delete);
indexUpdates.add(indexUpdate);
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
index e650fba..8b583f2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
@@ -29,6 +29,9 @@ import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.RowKeyColumnExpression;
+import org.apache.phoenix.expression.visitor.RowKeyExpressionVisitor;
import org.apache.phoenix.hbase.index.ValueGetter;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
@@ -182,7 +185,7 @@ public class IndexUtil {
}
};
- indexMutations.add(maintainer.buildUpdateMutation(kvBuilder, valueGetter, ptr, ts));
+ indexMutations.add(maintainer.buildUpdateMutation(kvBuilder, valueGetter, ptr, ts, null));
} else {
// We can only generate the correct Delete if we have no KV columns in our index.
// Perhaps it'd be best to ignore Delete mutations all together here, as this
@@ -211,4 +214,23 @@ public class IndexUtil {
}
return false;
}
+
+ /**
+ * Traverse the expression tree and set the offset of every RowKeyColumnExpression
+ * to the offset provided. This is used for local indexing on the server-side to
+ * skip over the region start key that prefixes index rows.
+ * @param rootExpression the root expression from which to begin traversal
+ * @param offset the offset to set on each RowKeyColumnExpression
+ */
+ public static void setRowKeyExpressionOffset(Expression rootExpression, final int offset) {
+ rootExpression.accept(new RowKeyExpressionVisitor() {
+
+ @Override
+ public Void visit(RowKeyColumnExpression node) {
+ node.setOffset(offset);
+ return null;
+ }
+
+ });
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
index be781c5..2e78022 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
@@ -34,7 +34,9 @@ import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
+import org.apache.phoenix.filter.BooleanExpressionFilter;
import org.apache.phoenix.filter.SkipScanFilter;
import org.apache.phoenix.query.KeyRange;
import org.apache.phoenix.query.KeyRange.Bound;
@@ -62,6 +64,14 @@ public class ScanUtil {
scan.setAttribute(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
}
+ public static void setLocalIndex(Scan scan) {
+ scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX, PDataType.TRUE_BYTES);
+ }
+
+ public static boolean isLocalIndex(Scan scan) {
+ return scan.getAttribute(BaseScannerRegionObserver.LOCAL_INDEX) != null;
+ }
+
// Use getTenantId and pass in column name to match against
// in as PSchema attribute. If column name matches in
// KeyExpressions, set on scan as attribute
@@ -414,4 +424,29 @@ public class ScanUtil {
byte[] reversed = scan.getAttribute(REVERSED_ATTR);
return (PDataType.TRUE_BYTES.equals(reversed));
}
+
+ private static void setRowKeyOffset(Filter filter, int offset) {
+ if (filter instanceof BooleanExpressionFilter) {
+ BooleanExpressionFilter boolFilter = (BooleanExpressionFilter)filter;
+ IndexUtil.setRowKeyExpressionOffset(boolFilter.getExpression(), offset);
+ } else if (filter instanceof SkipScanFilter) {
+ SkipScanFilter skipScanFilter = (SkipScanFilter)filter;
+ skipScanFilter.setOffset(offset);
+ }
+ }
+
+ public static void setRowKeyOffset(Scan scan, int offset) {
+ Filter filter = scan.getFilter();
+ if (filter == null) {
+ return;
+ }
+ if (filter instanceof FilterList) {
+ FilterList filterList = (FilterList)filter;
+ for (Filter childFilter : filterList.getFilters()) {
+ setRowKeyOffset(childFilter, offset);
+ }
+ } else {
+ setRowKeyOffset(filter, offset);
+ }
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/360f5a1c/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java b/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java
index 17d8b3f..42893ab 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/index/IndexMaintainerTest.java
@@ -132,7 +132,7 @@ public class IndexMaintainerTest extends BaseConnectionlessQueryTest {
ImmutableBytesWritable indexKeyPtr = new ImmutableBytesWritable(indexMutation.getRow());
ptr.set(rowKeyPtr.get(), rowKeyPtr.getOffset(), rowKeyPtr.getLength());
- byte[] mutablelndexRowKey = im1.buildRowKey(valueGetter, ptr);
+ byte[] mutablelndexRowKey = im1.buildRowKey(valueGetter, ptr, null);
byte[] immutableIndexRowKey = indexKeyPtr.copyBytes();
assertArrayEquals(immutableIndexRowKey, mutablelndexRowKey);