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:27 UTC
[12/50] incubator-kylin git commit: KYLIN-655, half way
KYLIN-655, half way
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/f037fe83
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/f037fe83
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/f037fe83
Branch: refs/heads/streaming-localdict
Commit: f037fe83ebfa50f861c5de205bb39801aaf6cf01
Parents: 5612b12
Author: Li, Yang <ya...@ebay.com>
Authored: Fri Apr 3 10:57:41 2015 +0800
Committer: Li, Yang <ya...@ebay.com>
Committed: Fri Apr 3 10:57:41 2015 +0800
----------------------------------------------------------------------
.../kylin/storage/cube/CubeGridTable.java | 2 -
.../storage/cube/CubeHBaseReadonlyStore.java | 82 +++++++++++++++-----
2 files changed, 64 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/f037fe83/storage/src/main/java/org/apache/kylin/storage/cube/CubeGridTable.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeGridTable.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeGridTable.java
index 641a269..3d3710f 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeGridTable.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeGridTable.java
@@ -19,7 +19,6 @@ import com.google.common.collect.Maps;
public class CubeGridTable {
- @SuppressWarnings("rawtypes")
public static Map<TblColRef, Dictionary<?>> getDimensionToDictionaryMap(CubeSegment cubeSeg, long cuboidId) {
CubeDesc cubeDesc = cubeSeg.getCubeDesc();
CubeManager cubeMgr = CubeManager.getInstance(cubeSeg.getCubeInstance().getConfig());
@@ -36,7 +35,6 @@ public class CubeGridTable {
return dictionaryMap;
}
- @SuppressWarnings("rawtypes")
public static GTInfo newGTInfo(CubeSegment cubeSeg, long cuboidId) {
Map<TblColRef, Dictionary<?>> dictionaryMap = getDimensionToDictionaryMap(cubeSeg, cuboidId);
return newGTInfo(cubeSeg.getCubeDesc(), cuboidId, dictionaryMap);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/f037fe83/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseReadonlyStore.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseReadonlyStore.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseReadonlyStore.java
index e1b8e3b..aff9125 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseReadonlyStore.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseReadonlyStore.java
@@ -2,12 +2,16 @@ package org.apache.kylin.storage.cube;
import java.io.IOException;
import java.util.BitSet;
+import java.util.Iterator;
+import java.util.List;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
import org.apache.kylin.common.persistence.HBaseConnection;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.cube.CubeSegment;
@@ -17,6 +21,7 @@ import org.apache.kylin.cube.model.HBaseColumnDesc;
import org.apache.kylin.cube.model.HBaseColumnFamilyDesc;
import org.apache.kylin.cube.model.HBaseMappingDesc;
import org.apache.kylin.storage.gridtable.GTInfo;
+import org.apache.kylin.storage.gridtable.GTRowBlock;
import org.apache.kylin.storage.gridtable.GTRowBlock.Writer;
import org.apache.kylin.storage.gridtable.GTScanRequest;
import org.apache.kylin.storage.gridtable.IGTStore;
@@ -24,8 +29,10 @@ import org.apache.kylin.storage.hbase.CubeSegmentTupleIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.Lists;
+
public class CubeHBaseReadonlyStore implements IGTStore {
-
+
public static final Logger logger = LoggerFactory.getLogger(CubeHBaseReadonlyStore.class);
public static final int SCAN_CACHE = 1024;
@@ -63,41 +70,80 @@ public class CubeHBaseReadonlyStore implements IGTStore {
@Override
public IGTStoreScanner scan(ByteArray pkStart, ByteArray pkEnd, BitSet selectedColBlocks, GTScanRequest additionalPushDown) throws IOException {
// TODO enable coprocessor
-
+
// globally shared connection, does not require close
HConnection hbaseConn = HBaseConnection.get(cubeSeg.getCubeInstance().getConfig().getStorageUrl());
- HTableInterface hbaseTable = hbaseConn.getTable(cubeSeg.getStorageLocationIdentifier());
-
- Scan hbaseScan = buildScan(pkStart, pkEnd, selectedColBlocks);
- ResultScanner scanner = hbaseTable.getScanner(hbaseScan);
-
-
-
- return null;
+ final HTableInterface hbaseTable = hbaseConn.getTable(cubeSeg.getStorageLocationIdentifier());
+ final List<Pair<byte[], byte[]>> hbaseColumns = makeHBaseColumns(selectedColBlocks);
+
+ Scan hbaseScan = buildScan(pkStart, pkEnd, hbaseColumns);
+ final ResultScanner scanner = hbaseTable.getScanner(hbaseScan);
+ final Iterator<Result> iterator = scanner.iterator();
+
+ return new IGTStoreScanner() {
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public GTRowBlock next() {
+ Result result = iterator.next();
+ result.getRow();
+ for (Pair<byte[], byte[]> hbaseColumn : hbaseColumns) {
+ result.getColumnLatestCell(hbaseColumn.getFirst(), hbaseColumn.getSecond());
+ }
+ return null;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws IOException {
+ scanner.close();
+ hbaseTable.close();
+ }
+ };
}
- private Scan buildScan(ByteArray pkStart, ByteArray pkEnd, BitSet selectedColBlocks) {
+ private Scan buildScan(ByteArray pkStart, ByteArray pkEnd, List<Pair<byte[], byte[]>> selectedColumns) {
Scan scan = new Scan();
scan.setCaching(SCAN_CACHE);
scan.setCacheBlocks(true);
scan.setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_ENABLE, Bytes.toBytes(Boolean.TRUE));
-
+
+ for (Pair<byte[], byte[]> hbaseColumn : selectedColumns) {
+ byte[] byteFamily = hbaseColumn.getFirst();
+ byte[] byteQualifier = hbaseColumn.getSecond();
+ scan.addColumn(byteFamily, byteQualifier);
+ }
+
+ scan.setStartRow(pkStart.copy().array());
+ scan.setStopRow(pkEnd.copy().array());
+ return scan;
+ }
+
+ private List<Pair<byte[], byte[]>> makeHBaseColumns(BitSet selectedColBlocks) {
+ List<Pair<byte[], byte[]>> result = Lists.newArrayList();
+
int colBlockIdx = 0;
HBaseMappingDesc hbaseMapping = cubeSeg.getCubeDesc().getHbaseMapping();
for (HBaseColumnFamilyDesc familyDesc : hbaseMapping.getColumnFamily()) {
+ byte[] byteFamily = Bytes.toBytes(familyDesc.getName());
for (HBaseColumnDesc hbaseColDesc : familyDesc.getColumns()) {
if (selectedColBlocks.get(colBlockIdx)) {
- byte[] byteFamily = Bytes.toBytes(familyDesc.getName());
byte[] byteQualifier = Bytes.toBytes(hbaseColDesc.getQualifier());
- scan.addColumn(byteFamily, byteQualifier);
+ result.add(new Pair<byte[], byte[]>(byteFamily, byteQualifier));
}
colBlockIdx++;
}
}
-
- scan.setStartRow(pkStart.copy().array());
- scan.setStopRow(pkEnd.copy().array());
- return scan;
+
+ return result;
}
}