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/03/21 01:27:07 UTC
[02/50] [abbrv] incubator-kylin git commit: refactor
refactor
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/5b3413b6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/5b3413b6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/5b3413b6
Branch: refs/heads/streaming-localdict
Commit: 5b3413b6f0b551f9ff19dea8cf1fb8425e63aac6
Parents: 8f03a27
Author: qianhao.zhou <qi...@ebay.com>
Authored: Wed Mar 18 14:30:08 2015 +0800
Committer: qianhao.zhou <qi...@ebay.com>
Committed: Wed Mar 18 14:30:08 2015 +0800
----------------------------------------------------------------------
.../apache/kylin/invertedindex/index/Slice.java | 2 +-
.../invertedindex/model/IIKeyValueCodec.java | 128 +------------------
.../hbase/coprocessor/endpoint/IIEndpoint.java | 12 +-
.../invertedindex/IIStreamBuilder.java | 11 +-
4 files changed, 22 insertions(+), 131 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/5b3413b6/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/Slice.java
----------------------------------------------------------------------
diff --git a/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/Slice.java b/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/Slice.java
index eb2fb92..7db0a93 100644
--- a/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/Slice.java
+++ b/invertedindex/src/main/java/org/apache/kylin/invertedindex/index/Slice.java
@@ -42,7 +42,7 @@ public class Slice implements Iterable<RawTableRecord>, Comparable<Slice> {
final short shard;
final long timestamp;
final int nRecords;
- ColumnValueContainer[] containers;
+ final ColumnValueContainer[] containers;
private Map<Integer, Dictionary<?>> localDictionaries;
public Slice(TableRecordInfoDigest digest, short shard, long timestamp, ColumnValueContainer[] containers) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/5b3413b6/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
----------------------------------------------------------------------
diff --git a/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java b/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
index e3284fc..d15f56b 100644
--- a/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
+++ b/invertedindex/src/main/java/org/apache/kylin/invertedindex/model/IIKeyValueCodec.java
@@ -65,7 +65,11 @@ public class IIKeyValueCodec implements KeyValueCodec {
ImmutableBytesWritable key = encodeKey(slice.getShard(), slice.getTimestamp(), col);
ImmutableBytesWritable value = container.toBytes();
final Dictionary<?> dictionary = slice.getLocalDictionaries().get(col);
- return new IIRow(key, value, serialize(dictionary));
+ if (dictionary == null) {
+ return new IIRow(key, value, new ImmutableBytesWritable(BytesUtil.EMPTY_BYTE_ARRAY));
+ } else {
+ return new IIRow(key, value, serialize(dictionary));
+ }
}
private static Dictionary<?> deserialize(ImmutableBytesWritable dictBytes) {
@@ -162,17 +166,17 @@ public class IIKeyValueCodec implements KeyValueCodec {
}
int curCol = BytesUtil.readUnsigned(key.get(), i, COLNO_LEN);
- final Dictionary<?> dictionary = deserialize(row.getDictionary());
if (incompleteDigest.isMetrics(curCol)) {
CompressedValueContainer c = new CompressedValueContainer(incompleteDigest, curCol, 0);
c.fromBytes(row.getValue());
valueContainers[curCol] = c;
} else {
+ final Dictionary<?> dictionary = deserialize(row.getDictionary());
CompressedValueContainer c = new CompressedValueContainer(dictionary.getSizeOfId(), dictionary.getMaxId() - dictionary.getMinId() + 1, 0);
c.fromBytes(row.getValue());
valueContainers[curCol] = c;
+ localDictionaries.put(curCol, dictionary);
}
- localDictionaries.put(curCol, dictionary);
columns++;
lastShard = curShard;
lastTimestamp = curTimestamp;
@@ -196,122 +200,4 @@ public class IIKeyValueCodec implements KeyValueCodec {
}
-
- private static class Decoder implements Iterable<Slice> {
-
- private final TableRecordInfoDigest digest;
- Iterator<IIRow> iterator;
-
- Slice slice = null;
- short curShard = Short.MIN_VALUE;
- long curSliceTimestamp = Long.MIN_VALUE;
- int curCol = -1;
- short lastShard = Short.MIN_VALUE;
- long lastSliceTimestamp = Long.MIN_VALUE;
- int lastCol = -1;
- ColumnValueContainer[] containers = null;
- Map<Integer, Dictionary<?>> localDictionaries = Maps.newHashMap();
-
- Decoder(Iterable<IIRow> kvs, TableRecordInfoDigest digest) {
- this.digest = digest;
- this.iterator = kvs.iterator();
- }
-
- private void goToNext() {
- if (slice != null) { // was not fetched
- return;
- }
-
- // NOTE the input keys are ordered
- while (slice == null && iterator.hasNext()) {
- IIRow kv = iterator.next();
- ImmutableBytesWritable k = kv.getKey();
- ImmutableBytesWritable v = kv.getValue();
- decodeKey(k);
- final Dictionary<?> dictionary = deserialize(kv.getDictionary());
- final CompressedValueContainer c = new CompressedValueContainer(digest, curCol, 0);
- c.fromBytes(kv.getValue());
- addContainer(curCol, c);
- localDictionaries.put(curCol, dictionary);
- if (localDictionaries.size() < digest.getColumnCount()) {
- continue;
- }
-
- if (curShard != lastShard
- || curSliceTimestamp != lastSliceTimestamp) {
- makeNext();
- }
- consumeCurrent(v);
- }
- if (slice == null) {
- makeNext();
- }
- }
-
- private void decodeKey(ImmutableBytesWritable k) {
- byte[] buf = k.get();
- int i = k.getOffset();
-
- curShard = (short) BytesUtil.readUnsigned(buf, i, SHARD_LEN);
- i += SHARD_LEN;
- curSliceTimestamp = BytesUtil.readLong(buf, i, TIMEPART_LEN);
- i += TIMEPART_LEN;
-
- curCol = BytesUtil.readUnsigned(buf, i, COLNO_LEN);
- i += COLNO_LEN;
-
- }
-
- private void consumeCurrent(ImmutableBytesWritable v) {
- lastShard = curShard;
- lastSliceTimestamp = curSliceTimestamp;
- lastCol = curCol;
- }
-
- private void makeNext() {
- if (containers != null) {
- slice = new Slice(digest, lastShard, lastSliceTimestamp,
- containers);
- slice.setLocalDictionaries(Maps.newHashMap(localDictionaries));
- }
- lastSliceTimestamp = Long.MIN_VALUE;
- lastCol = -1;
- containers = null;
- localDictionaries.clear();
- }
-
- private void addContainer(int col, ColumnValueContainer c) {
- if (containers == null) {
- containers = new ColumnValueContainer[digest.getColumnCount()];
- }
- containers[col] = c;
- }
-
- @Override
- public Iterator<Slice> iterator() {
- return new Iterator<Slice>() {
- @Override
- public boolean hasNext() {
- goToNext();
- return slice != null;
- }
-
- @Override
- public Slice next() {
- Slice result = slice;
- slice = null;
- return result;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
- };
- }
-
- }
-
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/5b3413b6/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 0e443c9..cb85718 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
@@ -30,6 +30,7 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
@@ -168,8 +169,15 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
byte[] result = new byte[digest.getByteFormLen()];
for (int i = 0; i < coprocessorRowType.getColumnCount(); i++) {
final TblColRef column = coprocessorRowType.columns[i];
- final int length = localDictionaries.get(i).getValueBytesFromId(encodedRecord.getValueID(i), buffer, 0);
- rowKeyColumnIO.writeColumn(column, buffer, length, Dictionary.NULL, result, digest.offset(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));
+ } else {
+ final int length = localDictionaries.get(i).getValueBytesFromId(encodedRecord.getValueID(i), buffer, 0);
+ rowKeyColumnIO.writeColumn(column, buffer, length, Dictionary.NULL, result, digest.offset(i));
+ }
}
return result;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/5b3413b6/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java b/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
index 480cb39..2ba90fd 100644
--- a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
+++ b/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
@@ -111,16 +111,13 @@ public class IIStreamBuilder extends StreamBuilder {
private Map<Integer, Dictionary<?>> buildDictionary(List<List<String>> table, IIDesc desc) {
HashMultimap<TblColRef, String> valueMap = HashMultimap.create();
- Set<TblColRef> dimensionColumns = Sets.newHashSet();
- for (int i = 0; i < desc.listAllColumns().size(); i++) {
- if (!desc.isMetricsCol(i)) {
- dimensionColumns.add(desc.listAllColumns().get(i));
- }
- }
+ final List<TblColRef> allColumns = desc.listAllColumns();
for (List<String> row : table) {
for (int i = 0; i < row.size(); i++) {
String cell = row.get(i);
- valueMap.put(desc.listAllColumns().get(i), cell);
+ if (!desc.isMetricsCol(i)) {
+ valueMap.put(allColumns.get(i), cell);
+ }
}
}
Map<Integer, Dictionary<?>> result = Maps.newHashMap();