You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2016/01/13 10:58:33 UTC
[9/9] kylin git commit: KYLIN-1308 maintain CubeHBaseScanRPC
KYLIN-1308 maintain CubeHBaseScanRPC
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/bc8ea781
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/bc8ea781
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/bc8ea781
Branch: refs/heads/2.x-staging
Commit: bc8ea78171856e6c438b1a5da5ed9e2b1a49a862
Parents: fe19e3d
Author: honma <ho...@ebay.com>
Authored: Wed Jan 13 17:42:54 2016 +0800
Committer: honma <ho...@ebay.com>
Committed: Wed Jan 13 17:54:04 2016 +0800
----------------------------------------------------------------------
.../apache/kylin/cube/kv/LazyRowKeyEncoder.java | 26 ++
.../storage/hbase/cube/v2/CubeHBaseRPC.java | 43 +++
.../storage/hbase/cube/v2/CubeHBaseScanRPC.java | 348 +++++++++++--------
.../hbase/cube/v2/CubeSegmentScanner.java | 2 +-
4 files changed, 266 insertions(+), 153 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/bc8ea781/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java
index 0fca726..c93f65e 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/LazyRowKeyEncoder.java
@@ -18,9 +18,15 @@
package org.apache.kylin.cube.kv;
+import com.google.common.collect.Lists;
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.common.util.ShardingHash;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
+import java.util.Arrays;
+import java.util.List;
+
/**
* A LazyRowKeyEncoder will not try to calculate shard
* It works for both enableSharding or non-enableSharding scenario
@@ -38,4 +44,24 @@ public class LazyRowKeyEncoder extends RowKeyEncoder {
throw new RuntimeException("If enableSharding false, you should never calculate shard");
}
}
+
+
+ //for non-sharding cases it will only return one byte[] with not shard at beginning
+ public List<byte[]> getRowKeysDifferentShards(byte[] halfCookedKey) {
+ final short cuboidShardNum = cubeSeg.getCuboidShardNum(cuboid.getId());
+
+ if (!enableSharding) {
+ return Lists.newArrayList(halfCookedKey);//not shard to append at head, so it is already well cooked
+ } else {
+ List<byte[]> ret = Lists.newArrayList();
+ for (short i = 0; i < cuboidShardNum; ++i) {
+ short shard = ShardingHash.normalize(cubeSeg.getCuboidBaseShard(cuboid.getId()), i, cubeSeg.getTotalShards());
+ byte[] cookedKey = Arrays.copyOf(halfCookedKey, halfCookedKey.length);
+ BytesUtil.writeShort(shard, cookedKey, 0, RowConstants.ROWKEY_SHARDID_LEN);
+ ret.add(cookedKey);
+ }
+ return ret;
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/bc8ea781/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
index 07143b1..db39455 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseRPC.java
@@ -30,6 +30,7 @@ import org.apache.kylin.gridtable.IGTScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
@@ -104,6 +105,48 @@ public abstract class CubeHBaseRPC {
return new RawScan(start, end, selectedColumns, hbaseFuzzyKeys, hbaseCaching, hbaseMaxResultSize);
}
+ protected List<RawScan> preparedHBaseScans(GTRecord pkStart, GTRecord pkEnd, List<GTRecord> fuzzyKeys, ImmutableBitSet selectedColBlocks) {
+ final List<Pair<byte[], byte[]>> selectedColumns = makeHBaseColumns(selectedColBlocks);
+ List<RawScan> ret = Lists.newArrayList();
+
+ LazyRowKeyEncoder encoder = new LazyRowKeyEncoder(cubeSeg, cuboid);
+ byte[] start = encoder.createBuf();
+ byte[] end = encoder.createBuf();
+ List<byte[]> startKeys;
+ List<byte[]> endKeys;
+
+ encoder.setBlankByte(RowConstants.ROWKEY_LOWER_BYTE);
+ encoder.encode(pkStart, pkStart.getInfo().getPrimaryKey(), start);
+ startKeys = encoder.getRowKeysDifferentShards(start);
+
+ encoder.setBlankByte(RowConstants.ROWKEY_UPPER_BYTE);
+ encoder.encode(pkEnd, pkEnd.getInfo().getPrimaryKey(), end);
+ endKeys = encoder.getRowKeysDifferentShards(end);
+ endKeys = Lists.transform(endKeys, new Function<byte[], byte[]>() {
+ @Override
+ public byte[] apply(byte[] input) {
+ byte[] shardEnd = new byte[input.length + 1];//append extra 0 to the end key to make it inclusive while scanning
+ System.arraycopy(input, 0, shardEnd, 0, input.length);
+ return shardEnd;
+ }
+ });
+
+ Preconditions.checkState(startKeys.size() == endKeys.size());
+ List<Pair<byte[], byte[]>> hbaseFuzzyKeys = translateFuzzyKeys(fuzzyKeys);
+
+ KylinConfig config = cubeSeg.getCubeDesc().getConfig();
+ int hbaseCaching = config.getHBaseScanCacheRows();
+ int hbaseMaxResultSize = config.getHBaseScanMaxResultSize();
+ if (isMemoryHungry(selectedColBlocks))
+ hbaseCaching /= 10;
+
+ for (short i = 0; i < startKeys.size(); ++i) {
+ ret.add(new RawScan(startKeys.get(i), endKeys.get(i), selectedColumns, hbaseFuzzyKeys, hbaseCaching, hbaseMaxResultSize));
+ }
+ return ret;
+
+ }
+
private boolean isMemoryHungry(ImmutableBitSet selectedColBlocks) {
ImmutableBitSet selectColumns = fullGTInfo.selectColumns(selectedColBlocks);
return fullGTInfo.getMaxColumnLength(selectColumns) > 1024;
http://git-wip-us.apache.org/repos/asf/kylin/blob/bc8ea781/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
index 90888a3..fa32769 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeHBaseScanRPC.java
@@ -1,152 +1,196 @@
-//package org.apache.kylin.storage.hbase.cube.v2;
-//
-//import java.io.IOException;
-//import java.util.Iterator;
-//import java.util.List;
-//
-//import org.apache.hadoop.hbase.Cell;
-//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.kylin.common.util.ImmutableBitSet;
-//import org.apache.kylin.cube.CubeSegment;
-//import org.apache.kylin.cube.cuboid.Cuboid;
-//import org.apache.kylin.gridtable.GTInfo;
-//import org.apache.kylin.gridtable.GTRecord;
-//import org.apache.kylin.gridtable.GTScanRequest;
-//import org.apache.kylin.gridtable.IGTScanner;
-//import org.apache.kylin.gridtable.IGTStore;
-//import org.apache.kylin.storage.hbase.HBaseConnection;
-//import org.slf4j.Logger;
-//import org.slf4j.LoggerFactory;
-//
-//import com.google.common.collect.Iterators;
-//import com.google.common.collect.Lists;
-//
-///**
-// * for test use only
-// */
-//public class CubeHBaseScanRPC extends CubeHBaseRPC {
-// public static final Logger logger = LoggerFactory.getLogger(CubeHBaseScanRPC.class);
-//
-// static class TrimmedInfoGTRecordAdapter implements Iterable<GTRecord> {
-//
-// private final GTInfo info;
-// private final Iterator<GTRecord> input;
-//
-// public TrimmedInfoGTRecordAdapter(GTInfo info, Iterator<GTRecord> input) {
-// this.info = info;
-// this.input = input;
-// }
-//
-// @Override
-// public Iterator<GTRecord> iterator() {
-// return new Iterator<GTRecord>() {
-// @Override
-// public boolean hasNext() {
-// return input.hasNext();
-// }
-//
-// @Override
-// public GTRecord next() {
-// GTRecord x = input.next();
-// return new GTRecord(info, x.getInternal());
-// }
-//
-// @Override
-// public void remove() {
-//
-// }
-// };
-// }
-// }
-//
-// public CubeHBaseScanRPC(CubeSegment cubeSeg, Cuboid cuboid, GTInfo fullGTInfo) {
-// super(cubeSeg, cuboid, fullGTInfo);
-// }
-//
-// @Override
-// public IGTScanner getGTScanner(final GTScanRequest scanRequest) throws IOException {
-//
-// // primary key (also the 0th column block) is always selected
-// final ImmutableBitSet selectedColBlocks = scanRequest.getSelectedColBlocks().set(0);
-// // globally shared connection, does not require close
-// HConnection hbaseConn = HBaseConnection.get(cubeSeg.getCubeInstance().getConfig().getStorageUrl());
-// final HTableInterface hbaseTable = hbaseConn.getTable(cubeSeg.getStorageLocationIdentifier());
-//
-// List<RawScan> rawScans = preparedHBaseScan(scanRequest.getPkStart(), scanRequest.getPkEnd(), scanRequest.getFuzzyKeys(), selectedColBlocks);
-// List<List<Integer>> hbaseColumnsToGT = getHBaseColumnsGTMapping(selectedColBlocks);
-//
-// final List<ResultScanner> scanners = Lists.newArrayList();
-// final List<Iterator<Result>> resultIterators = Lists.newArrayList();
-//
-// for (RawScan rawScan : rawScans) {
-//
-// logScan(rawScan, cubeSeg.getStorageLocationIdentifier());
-// Scan hbaseScan = buildScan(rawScan);
-//
-// final ResultScanner scanner = hbaseTable.getScanner(hbaseScan);
-// final Iterator<Result> iterator = scanner.iterator();
-//
-// scanners.add(scanner);
-// resultIterators.add(iterator);
-// }
-//
-// final Iterator<Result> allResultsIterator = Iterators.concat(resultIterators.iterator());
-//
-// CellListIterator cellListIterator = new CellListIterator() {
-// @Override
-// public void close() throws IOException {
-// for (ResultScanner scanner : scanners) {
-// scanner.close();
-// }
-// hbaseTable.close();
-// }
-//
-// @Override
-// public boolean hasNext() {
-// return allResultsIterator.hasNext();
-// }
-//
-// @Override
-// public List<Cell> next() {
-// return allResultsIterator.next().listCells();
-// }
-//
-// @Override
-// public void remove() {
-// throw new UnsupportedOperationException();
-// }
-// };
-//
-// IGTStore store = new HBaseReadonlyStore(cellListIterator, scanRequest, rawScans.get(0).hbaseColumns, hbaseColumnsToGT, cubeSeg.getRowKeyPreambleSize());
-// IGTScanner rawScanner = store.scan(scanRequest);
-//
-// final IGTScanner decorateScanner = scanRequest.decorateScanner(rawScanner);
-// final TrimmedInfoGTRecordAdapter trimmedInfoGTRecordAdapter = new TrimmedInfoGTRecordAdapter(fullGTInfo, decorateScanner.iterator());
-//
-// return new IGTScanner() {
-// @Override
-// public GTInfo getInfo() {
-// return fullGTInfo;
-// }
-//
-// @Override
-// public int getScannedRowCount() {
-// return decorateScanner.getScannedRowCount();
-// }
-//
-// @Override
-// public void close() throws IOException {
-// decorateScanner.close();
-// }
-//
-// @Override
-// public Iterator<GTRecord> iterator() {
-// return trimmedInfoGTRecordAdapter.iterator();
-// }
-// };
-// }
-//}
+package org.apache.kylin.storage.hbase.cube.v2;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.hbase.Cell;
+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.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRecord;
+import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.gridtable.IGTScanner;
+import org.apache.kylin.gridtable.IGTStore;
+import org.apache.kylin.storage.hbase.HBaseConnection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+
+/**
+ * for test use only
+ */
+public class CubeHBaseScanRPC extends CubeHBaseRPC {
+ public static final Logger logger = LoggerFactory.getLogger(CubeHBaseScanRPC.class);
+
+ static class TrimmedInfoGTRecordAdapter implements Iterable<GTRecord> {
+
+ private final GTInfo info;
+ private final Iterator<GTRecord> input;
+
+ public TrimmedInfoGTRecordAdapter(GTInfo info, Iterator<GTRecord> input) {
+ this.info = info;
+ this.input = input;
+ }
+
+ @Override
+ public Iterator<GTRecord> iterator() {
+ return new Iterator<GTRecord>() {
+ @Override
+ public boolean hasNext() {
+ return input.hasNext();
+ }
+
+ @Override
+ public GTRecord next() {
+ GTRecord x = input.next();
+ return new GTRecord(info, x.getInternal());
+ }
+
+ @Override
+ public void remove() {
+
+ }
+ };
+ }
+ }
+
+ public CubeHBaseScanRPC(CubeSegment cubeSeg, Cuboid cuboid, GTInfo fullGTInfo) {
+ super(cubeSeg, cuboid, fullGTInfo);
+ }
+
+ @Override
+ public IGTScanner getGTScanner(final List<GTScanRequest> scanRequests) throws IOException {
+ final List<IGTScanner> scanners = Lists.newArrayList();
+ for (GTScanRequest request : scanRequests) {
+ scanners.add(getGTScanner(request));
+ }
+
+ return new IGTScanner() {
+ @Override
+ public GTInfo getInfo() {
+ return scanners.get(0).getInfo();
+ }
+
+ @Override
+ public int getScannedRowCount() {
+ int sum = 0;
+ for (IGTScanner s : scanners) {
+ sum += s.getScannedRowCount();
+ }
+ return sum;
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (IGTScanner s : scanners) {
+ s.close();
+ }
+ }
+
+ @Override
+ public Iterator<GTRecord> iterator() {
+ return Iterators.concat(Iterators.transform(scanners.iterator(), new Function<IGTScanner, Iterator<GTRecord>>() {
+ @Nullable
+ @Override
+ public Iterator<GTRecord> apply(IGTScanner input) {
+ return input.iterator();
+ }
+ }));
+ }
+ };
+ }
+
+ private IGTScanner getGTScanner(final GTScanRequest scanRequest) throws IOException {
+
+ // primary key (also the 0th column block) is always selected
+ final ImmutableBitSet selectedColBlocks = scanRequest.getSelectedColBlocks().set(0);
+ // globally shared connection, does not require close
+ HConnection hbaseConn = HBaseConnection.get(cubeSeg.getCubeInstance().getConfig().getStorageUrl());
+ final HTableInterface hbaseTable = hbaseConn.getTable(cubeSeg.getStorageLocationIdentifier());
+
+ List<RawScan> rawScans = preparedHBaseScans(scanRequest.getPkStart(), scanRequest.getPkEnd(), scanRequest.getFuzzyKeys(), selectedColBlocks);
+ List<List<Integer>> hbaseColumnsToGT = getHBaseColumnsGTMapping(selectedColBlocks);
+
+ final List<ResultScanner> scanners = Lists.newArrayList();
+ final List<Iterator<Result>> resultIterators = Lists.newArrayList();
+
+ for (RawScan rawScan : rawScans) {
+
+ logScan(rawScan, cubeSeg.getStorageLocationIdentifier());
+ Scan hbaseScan = buildScan(rawScan);
+
+ final ResultScanner scanner = hbaseTable.getScanner(hbaseScan);
+ final Iterator<Result> iterator = scanner.iterator();
+
+ scanners.add(scanner);
+ resultIterators.add(iterator);
+ }
+
+ final Iterator<Result> allResultsIterator = Iterators.concat(resultIterators.iterator());
+
+ CellListIterator cellListIterator = new CellListIterator() {
+ @Override
+ public void close() throws IOException {
+ for (ResultScanner scanner : scanners) {
+ scanner.close();
+ }
+ hbaseTable.close();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return allResultsIterator.hasNext();
+ }
+
+ @Override
+ public List<Cell> next() {
+ return allResultsIterator.next().listCells();
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+
+ IGTStore store = new HBaseReadonlyStore(cellListIterator, scanRequest, rawScans.get(0).hbaseColumns, hbaseColumnsToGT, cubeSeg.getRowKeyPreambleSize());
+ IGTScanner rawScanner = store.scan(scanRequest);
+
+ final IGTScanner decorateScanner = scanRequest.decorateScanner(rawScanner);
+ final TrimmedInfoGTRecordAdapter trimmedInfoGTRecordAdapter = new TrimmedInfoGTRecordAdapter(fullGTInfo, decorateScanner.iterator());
+
+ return new IGTScanner() {
+ @Override
+ public GTInfo getInfo() {
+ return fullGTInfo;
+ }
+
+ @Override
+ public int getScannedRowCount() {
+ return decorateScanner.getScannedRowCount();
+ }
+
+ @Override
+ public void close() throws IOException {
+ decorateScanner.close();
+ }
+
+ @Override
+ public Iterator<GTRecord> iterator() {
+ return trimmedInfoGTRecordAdapter.iterator();
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/bc8ea781/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
index 5798f41..ad0cb32 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeSegmentScanner.java
@@ -225,7 +225,7 @@ public class CubeSegmentScanner implements IGTScanner {
public Scanner() {
CubeHBaseRPC rpc;
if ("scan".equalsIgnoreCase(BackdoorToggles.getHbaseCubeQueryProtocol())) {
- rpc = null;
+ rpc = new CubeHBaseScanRPC(cubeSeg, cuboid, info);
} else {
rpc = new CubeHBaseEndpointRPC(cubeSeg, cuboid, info);//default behavior
}