You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/04/03 15:03:43 UTC
[28/50] incubator-kylin git commit: optimize Endpoint
optimize Endpoint
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/4f40ca4e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/4f40ca4e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/4f40ca4e
Branch: refs/heads/streaming-localdict
Commit: 4f40ca4e034f112598190012c3304c949b84800e
Parents: dc441a6
Author: qianhao.zhou <qi...@ebay.com>
Authored: Fri Apr 3 15:44:38 2015 +0800
Committer: qianhao.zhou <qi...@ebay.com>
Committed: Fri Apr 3 15:44:38 2015 +0800
----------------------------------------------------------------------
.../invertedindex/index/RawTableRecord.java | 8 +++--
.../hbase/coprocessor/endpoint/IIEndpoint.java | 33 ++++++++++----------
2 files changed, 23 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4f40ca4e/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/RawTableRecord.java
----------------------------------------------------------------------
diff --git a/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/RawTableRecord.java b/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/RawTableRecord.java
index b7577c0..c300fc5 100644
--- a/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/RawTableRecord.java
+++ b/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/RawTableRecord.java
@@ -57,10 +57,14 @@ public class RawTableRecord implements Cloneable {
return digest.codec(col);
}
- public int length(int col) {
+ public final int length(int col) {
return digest.length(col);
}
+ public final int offset(int col) {
+ return digest.offset(col);
+ }
+
public int getColumnCount() {
return digest.getColumnCount();
}
@@ -97,7 +101,7 @@ public class RawTableRecord implements Cloneable {
}
public void getValueBytes(int col, ImmutableBytesWritable bytes) {
- bytes.set(buf, digest.offset(col), digest.length(col));
+ bytes.set(buf, offset(col), length(col));
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4f40ca4e/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
index c13a841..826a802 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
@@ -136,6 +136,7 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
final byte[] buffer = new byte[CoprocessorConstants.METRIC_SERIALIZE_BUFFER_SIZE];
ClearTextDictionary clearTextDictionary = new ClearTextDictionary(recordInfo, type);
RowKeyColumnIO rowKeyColumnIO = new RowKeyColumnIO(clearTextDictionary);
+ byte[] recordBuffer = new byte[recordInfo.getByteFormLen()];
for (Slice slice : slices) {
//TODO localdict
@@ -152,10 +153,10 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
Iterator<RawTableRecord> iterator = slice.iterateWithBitmap(result);
while (iterator.hasNext()) {
final RawTableRecord rawTableRecord = iterator.next();
- byte[] data = decodeWithDictionary(rawTableRecord, localDictionaries, recordInfo, buffer, rowKeyColumnIO, type);
- CoprocessorProjector.AggrKey aggKey = projector.getAggrKey(data);
+ decodeWithDictionary(recordBuffer, rawTableRecord, localDictionaries, recordInfo, buffer, rowKeyColumnIO, type);
+ CoprocessorProjector.AggrKey aggKey = projector.getAggrKey(recordBuffer);
MeasureAggregator[] bufs = aggCache.getBuffer(aggKey);
- aggregators.aggregate(bufs, data);
+ aggregators.aggregate(bufs, recordBuffer);
aggCache.checkMemoryUsage();
}
}
@@ -171,21 +172,20 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
return responseBuilder.build();
}
- private byte[] decodeWithDictionary(RawTableRecord encodedRecord, Map<Integer, Dictionary<?>> localDictionaries, TableRecordInfoDigest digest, byte[] buffer, RowKeyColumnIO rowKeyColumnIO, CoprocessorRowType coprocessorRowType) {
- byte[] result = new byte[digest.getByteFormLen()];
- for (int i = 0; i < coprocessorRowType.getColumnCount(); i++) {
- final TblColRef column = coprocessorRowType.columns[i];
- if (digest.isMetrics(i)) {
- final ImmutableBytesWritable bytes = new ImmutableBytesWritable();
- encodedRecord.getValueBytes(i, bytes);
- System.arraycopy(bytes.get(), bytes.getOffset(), buffer, 0, bytes.getLength());
- rowKeyColumnIO.writeColumn(column, buffer, bytes.getLength(), Dictionary.NULL, result, digest.offset(i));
+ private void decodeWithDictionary(byte[] recordBuffer, RawTableRecord encodedRecord, Map<Integer, Dictionary<?>> localDictionaries, TableRecordInfoDigest digest, byte[] buffer, RowKeyColumnIO rowKeyColumnIO, CoprocessorRowType coprocessorRowType) {
+ final TblColRef[] columns = coprocessorRowType.columns;
+ final int columnSize = columns.length;
+ final boolean[] isMetric = digest.isMetrics();
+ for (int i = 0; i < columnSize; i++) {
+ final TblColRef column = columns[i];
+ if (isMetric[i]) {
+ System.arraycopy(encodedRecord.getBytes(), encodedRecord.offset(i), buffer, 0, encodedRecord.length(i));
+ rowKeyColumnIO.writeColumn(column, buffer, encodedRecord.length(i), Dictionary.NULL, recordBuffer, digest.offset(i));
} else {
final int length = localDictionaries.get(i).getValueBytesFromId(encodedRecord.getValueID(i), buffer, 0);
- rowKeyColumnIO.writeColumn(column, buffer, length, Dictionary.NULL, result, digest.offset(i));
+ rowKeyColumnIO.writeColumn(column, buffer, length, Dictionary.NULL, recordBuffer, digest.offset(i));
}
}
- return result;
}
private IIProtos.IIResponse getNonAggregatedResponse(Iterable<Slice> slices, TableRecordInfoDigest recordInfo, CoprocessorFilter filter, CoprocessorRowType type) {
@@ -193,6 +193,7 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
final byte[] buffer = new byte[CoprocessorConstants.METRIC_SERIALIZE_BUFFER_SIZE];
ClearTextDictionary clearTextDictionary = new ClearTextDictionary(recordInfo, type);
RowKeyColumnIO rowKeyColumnIO = new RowKeyColumnIO(clearTextDictionary);
+ byte[] recordBuffer = new byte[recordInfo.getByteFormLen()];
for (Slice slice : slices) {
CoprocessorFilter newFilter = CoprocessorFilter.fromFilter(new LocalDictionary(slice.getLocalDictionaries(), type, slice.getInfo()), filter.getFilter(), FilterDecorator.FilterConstantsTreatment.REPLACE_WITH_LOCAL_DICT);
ConciseSet result = null;
@@ -203,8 +204,8 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
Iterator<RawTableRecord> iterator = slice.iterateWithBitmap(result);
while (iterator.hasNext()) {
final RawTableRecord rawTableRecord = iterator.next();
- byte[] data = decodeWithDictionary(rawTableRecord, slice.getLocalDictionaries(), recordInfo, buffer, rowKeyColumnIO, type);
- IIProtos.IIResponse.IIRow.Builder rowBuilder = IIProtos.IIResponse.IIRow.newBuilder().setColumns(ByteString.copyFrom(data));
+ decodeWithDictionary(recordBuffer, rawTableRecord, slice.getLocalDictionaries(), recordInfo, buffer, rowKeyColumnIO, type);
+ IIProtos.IIResponse.IIRow.Builder rowBuilder = IIProtos.IIResponse.IIRow.newBuilder().setColumns(ByteString.copyFrom(recordBuffer));
responseBuilder.addRows(rowBuilder.build());
}
}