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 2015/08/13 12:41:41 UTC
[3/4] incubator-kylin git commit: KYLIN-909 Adapt GTStore to hbase
endpoint
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java b/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
index 77e3010..f3dd900 100644
--- a/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
+++ b/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
@@ -29,7 +29,7 @@ import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.storage.cache.CacheFledgedDynamicStorageEngine;
import org.apache.kylin.storage.cache.CacheFledgedStaticStorageEngine;
-import org.apache.kylin.storage.hbase.CubeStorageQuery;
+import org.apache.kylin.storage.cube.CubeStorageQuery;
import org.apache.kylin.storage.hbase.InvertedIndexStorageQuery;
import org.apache.kylin.storage.hybrid.HybridInstance;
import org.apache.kylin.storage.hybrid.HybridStorageEngine;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/CellListIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CellListIterator.java b/storage/src/main/java/org/apache/kylin/storage/cube/CellListIterator.java
new file mode 100644
index 0000000..d2c4191
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CellListIterator.java
@@ -0,0 +1,28 @@
+/*
+ * 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.kylin.storage.cube;
+
+import java.io.Closeable;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+
+public interface CellListIterator extends Iterator<List<Cell>>, Closeable{
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java
new file mode 100644
index 0000000..4c24d76
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java
@@ -0,0 +1,191 @@
+/*
+ * 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.kylin.storage.cube;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.zip.DataFormatException;
+
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.hbase.client.HConnection;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.coprocessor.Batch;
+import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
+import org.apache.hadoop.hbase.ipc.ServerRpcController;
+import org.apache.kylin.common.util.CompressionUtils;
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.storage.util.KryoUtils;
+import org.apache.kylin.common.util.Pair;
+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.storage.hbase.HBaseConnection;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.Builder;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitResponse;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitService;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Collections2;
+import com.google.common.collect.Iterators;
+import com.google.protobuf.ByteString;
+
+public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
+
+ static class EndpintResultsAsGTScanner implements IGTScanner {
+ private GTInfo info;
+ private Iterator<byte[]> blocks;
+
+ private int scannedRowCount = 0;
+ private int scannedRowBlockCount = 0;
+
+ public EndpintResultsAsGTScanner(GTInfo info, Iterator<byte[]> blocks) {
+ this.info = info;
+ this.blocks = blocks;
+ }
+
+ @Override
+ public GTInfo getInfo() {
+ return info;
+ }
+
+ @Override
+ public int getScannedRowCount() {
+ return 0;
+ }
+
+ @Override
+ public int getScannedRowBlockCount() {
+ return 0;
+ }
+
+ @Override
+ public void close() throws IOException {
+ //do nothing
+ }
+
+ @Override
+ public Iterator<GTRecord> iterator() {
+ return Iterators.concat(Iterators.transform(blocks, new Function<byte[], Iterator<GTRecord>>() {
+ @Nullable
+ @Override
+ public Iterator<GTRecord> apply(@Nullable final byte[] input) {
+
+ logger.info("Reassembling a raw block returned from Endpoint with byte length: " + input.length);
+ return new Iterator<GTRecord>() {
+ private ByteBuffer inputBuffer = null;
+ private GTRecord oneRecord = null;
+
+ @Override
+ public boolean hasNext() {
+ if (inputBuffer == null) {
+ inputBuffer = ByteBuffer.wrap(input);
+ oneRecord = new GTRecord(info);
+ }
+
+ return inputBuffer.position() < inputBuffer.limit();
+ }
+
+ @Override
+ public GTRecord next() {
+ oneRecord.loadAllColumns(inputBuffer);
+ return oneRecord;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ }));
+ }
+ }
+
+ public CubeHBaseEndpointRPC(CubeSegment cubeSeg, Cuboid cuboid, GTInfo fullGTInfo) {
+ super(cubeSeg, cuboid, fullGTInfo);
+ }
+
+ @Override
+ public IGTScanner getGTScanner(final GTScanRequest scanRequest) throws IOException {
+
+ try {
+ // 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());
+ final List<Pair<byte[], byte[]>> hbaseColumns = makeHBaseColumns(selectedColBlocks);
+
+ RawScan rawScan = prepareRawScan(scanRequest.getPkStart(), scanRequest.getPkEnd(), hbaseColumns);
+
+ byte[] scanRequestBytes = KryoUtils.serialize(scanRequest);
+ byte[] rawScanBytes = KryoUtils.serialize(rawScan);
+ Builder builder = CubeVisitRequest.newBuilder();
+ builder.setGtScanRequest(ByteString.copyFrom(scanRequestBytes)).setHbaseRawScan(ByteString.copyFrom(rawScanBytes));
+
+ Collection<CubeVisitResponse> results = getResults(builder.build(), hbaseTable, rawScan.startKey, rawScan.endKey);
+ final Collection<byte[]> rowBlocks = Collections2.transform(results, new Function<CubeVisitResponse, byte[]>() {
+ @Nullable
+ @Override
+ public byte[] apply(CubeVisitResponse input) {
+ try {
+ return CompressionUtils.decompress(input.getCompressedRows().toByteArray());
+ } catch (IOException | DataFormatException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+
+ return new EndpintResultsAsGTScanner(fullGTInfo, rowBlocks.iterator());
+
+ } catch (Throwable throwable) {
+ throwable.printStackTrace();
+ }
+ return null;
+ }
+
+ //TODO : async callback
+ private Collection<CubeVisitResponse> getResults(final CubeVisitRequest request, HTableInterface table, byte[] startKey, byte[] endKey) throws Throwable {
+ Map<byte[], CubeVisitResponse> results = table.coprocessorService(CubeVisitService.class, startKey, endKey, new Batch.Call<CubeVisitService, CubeVisitResponse>() {
+ public CubeVisitResponse call(CubeVisitService rowsService) throws IOException {
+ ServerRpcController controller = new ServerRpcController();
+ BlockingRpcCallback<CubeVisitResponse> rpcCallback = new BlockingRpcCallback<>();
+ rowsService.visitCube(controller, request, rpcCallback);
+ CubeVisitResponse response = rpcCallback.get();
+ if (controller.failedOnException()) {
+ throw controller.getFailedOn();
+ }
+ return response;
+ }
+ });
+
+ logger.info("{} regions returned results ", results.values().size());
+
+ return results.values();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
new file mode 100644
index 0000000..d3c7b0c
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
@@ -0,0 +1,160 @@
+package org.apache.kylin.storage.cube;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.FuzzyRowFilter;
+import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.Bytes;
+import org.apache.kylin.common.util.BytesUtil;
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.cube.kv.RowConstants;
+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.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRecord;
+import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.gridtable.IGTScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+public abstract class CubeHBaseRPC {
+
+ public static final Logger logger = LoggerFactory.getLogger(CubeHBaseRPC.class);
+
+ public static final int SCAN_CACHE = 1024;
+
+ final protected CubeSegment cubeSeg;
+ final protected Cuboid cuboid;
+ final protected GTInfo fullGTInfo;
+
+ public CubeHBaseRPC(CubeSegment cubeSeg, Cuboid cuboid,GTInfo fullGTInfo) {
+ this.cubeSeg = cubeSeg;
+ this.cuboid = cuboid;
+ this.fullGTInfo = fullGTInfo;
+ }
+
+ abstract IGTScanner getGTScanner(GTScanRequest scanRequest) throws IOException;
+
+ public static Scan buildScan(RawScan rawScan) {
+ Scan scan = new Scan();
+ scan.setCaching(SCAN_CACHE);
+ scan.setCacheBlocks(true);
+ scan.setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_ENABLE, Bytes.toBytes(Boolean.TRUE));
+
+ if (rawScan.startKey != null) {
+ scan.setStartRow(rawScan.startKey);
+ }
+ if (rawScan.endKey != null) {
+ scan.setStopRow(rawScan.endKey);
+ }
+ if (rawScan.fuzzyKey != null) {
+ applyFuzzyFilter(scan, rawScan.fuzzyKey);
+ }
+ if (rawScan.hbaseColumns != null) {
+ applyHBaseColums(scan, rawScan.hbaseColumns);
+ }
+
+ return scan;
+ }
+
+ protected RawScan prepareRawScan(GTRecord pkStart, GTRecord pkEnd, List<Pair<byte[], byte[]>> selectedColumns) {
+ byte[] start = makeRowKeyToScan(pkStart, (byte) 0x00);
+ byte[] end = makeRowKeyToScan(pkEnd, (byte) 0xff);
+
+ //TODO fuzzy match
+
+ return new RawScan(start, end, selectedColumns, null);
+ }
+
+ private byte[] makeRowKeyToScan(GTRecord pkRec, byte fill) {
+ ByteArray pk = GTRecord.exportScanKey(pkRec);
+ int pkMaxLen = pkRec.getInfo().getMaxColumnLength(pkRec.getInfo().getPrimaryKey());
+
+ byte[] buf = new byte[pkMaxLen + RowConstants.ROWKEY_CUBOIDID_LEN];
+ Arrays.fill(buf, fill);
+
+ System.arraycopy(cuboid.getBytes(), 0, buf, 0, RowConstants.ROWKEY_CUBOIDID_LEN);
+ if (pk != null && pk.array() != null) {
+ System.arraycopy(pk.array(), pk.offset(), buf, RowConstants.ROWKEY_CUBOIDID_LEN, pk.length());
+ }
+ return buf;
+ }
+
+ protected List<Pair<byte[], byte[]>> makeHBaseColumns(ImmutableBitSet selectedColBlocks) {
+ List<Pair<byte[], byte[]>> result = Lists.newArrayList();
+
+ int colBlockIdx = 1; // start from 1; the 0th column block is primary key which maps to rowkey
+ 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[] byteQualifier = Bytes.toBytes(hbaseColDesc.getQualifier());
+ result.add(new Pair<byte[], byte[]>(byteFamily, byteQualifier));
+ }
+ colBlockIdx++;
+ }
+ }
+
+ return result;
+ }
+
+ //possible to use binary search as cells might be sorted
+ public static Cell findCell(List<Cell> cells, byte[] familyName, byte[] columnName) {
+ for (Cell c : cells) {
+ if (BytesUtil.compareBytes(familyName, 0, c.getFamilyArray(), c.getFamilyOffset(), familyName.length) == 0 && //
+ BytesUtil.compareBytes(columnName, 0, c.getQualifierArray(), c.getQualifierOffset(), columnName.length) == 0) {
+ return c;
+ }
+ }
+ return null;
+ }
+
+ public static void applyHBaseColums(Scan scan, List<Pair<byte[], byte[]>> hbaseColumns) {
+ for (Pair<byte[], byte[]> hbaseColumn : hbaseColumns) {
+ byte[] byteFamily = hbaseColumn.getFirst();
+ byte[] byteQualifier = hbaseColumn.getSecond();
+ scan.addColumn(byteFamily, byteQualifier);
+ }
+ }
+
+ public static void applyFuzzyFilter(Scan scan, List<org.apache.kylin.common.util.Pair<byte[], byte[]>> fuzzyKeys) {
+ if (fuzzyKeys != null && fuzzyKeys.size() > 0) {
+ FuzzyRowFilter rowFilter = new FuzzyRowFilter(convertToHBasePair(fuzzyKeys));
+
+ Filter filter = scan.getFilter();
+ if (filter != null) {
+ // may have existed InclusiveStopFilter, see buildScan
+ FilterList filterList = new FilterList();
+ filterList.addFilter(filter);
+ filterList.addFilter(rowFilter);
+ scan.setFilter(filterList);
+ } else {
+ scan.setFilter(rowFilter);
+ }
+ }
+ }
+
+ private static List<org.apache.hadoop.hbase.util.Pair<byte[], byte[]>> convertToHBasePair(List<org.apache.kylin.common.util.Pair<byte[], byte[]>> pairList) {
+ List<org.apache.hadoop.hbase.util.Pair<byte[], byte[]>> result = Lists.newArrayList();
+ for (org.apache.kylin.common.util.Pair pair : pairList) {
+ org.apache.hadoop.hbase.util.Pair element = new org.apache.hadoop.hbase.util.Pair(pair.getFirst(), pair.getSecond());
+ result.add(element);
+ }
+
+ return result;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/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
deleted file mode 100644
index 8001fbd..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseReadonlyStore.java
+++ /dev/null
@@ -1,177 +0,0 @@
-package org.apache.kylin.storage.cube;
-
-import java.io.IOException;
-import java.util.Arrays;
-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.ByteArray;
-import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.common.util.ImmutableBitSet;
-import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.kv.RowConstants;
-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.gridtable.GTInfo;
-import org.apache.kylin.gridtable.GTRecord;
-import org.apache.kylin.gridtable.GTRowBlock;
-import org.apache.kylin.gridtable.GTScanRequest;
-import org.apache.kylin.gridtable.IGTStore;
-import org.apache.kylin.gridtable.GTRowBlock.Writer;
-import org.apache.kylin.storage.hbase.HBaseConnection;
-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;
-
- final private GTInfo info;
- final private CubeSegment cubeSeg;
- final private Cuboid cuboid;
-
- public CubeHBaseReadonlyStore(GTInfo info, CubeSegment cubeSeg, Cuboid cuboid) {
- this.info = info;
- this.cubeSeg = cubeSeg;
- this.cuboid = cuboid;
- }
-
- @Override
- public GTInfo getInfo() {
- return info;
- }
-
- @Override
- public IGTStoreWriter rebuild(int shard) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public IGTStoreWriter append(int shard, Writer fillLast) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public IGTStoreScanner scan(GTRecord pkStart, GTRecord pkEnd, ImmutableBitSet selectedColumnBlocks, GTScanRequest additionalPushDown) throws IOException {
- // TODO enable coprocessor
-
- // primary key (also the 0th column block) is always selected
- final ImmutableBitSet selectedColBlocks = selectedColumnBlocks.set(0);
-
- // globally shared connection, does not require close
- HConnection hbaseConn = HBaseConnection.get(cubeSeg.getCubeInstance().getConfig().getStorageUrl());
-
- 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();
- final GTRowBlock oneBlock = new GTRowBlock(info); // avoid object creation
- oneBlock.setNumberOfRows(1); // row block is always disabled for cubes, row block contains only one record
-
- return new IGTStoreScanner() {
-
- @Override
- public boolean hasNext() {
- return iterator.hasNext();
- }
-
- @Override
- public GTRowBlock next() {
- Result result = iterator.next();
-
- // dimensions, set to primary key, also the 0th column block
- byte[] rowkey = result.getRow();
- oneBlock.getPrimaryKey().set(rowkey, RowConstants.ROWKEY_CUBOIDID_LEN, rowkey.length - RowConstants.ROWKEY_CUBOIDID_LEN);
- oneBlock.getCellBlock(0).set(rowkey, RowConstants.ROWKEY_CUBOIDID_LEN, rowkey.length - RowConstants.ROWKEY_CUBOIDID_LEN);
-
- // metrics
- int hbaseColIdx = 0;
- for (int i = 0; i < selectedColBlocks.trueBitCount(); i++) {
- int colBlockIdx = selectedColBlocks.trueBitAt(i);
- Pair<byte[], byte[]> hbaseColumn = hbaseColumns.get(hbaseColIdx++);
- Cell cell = result.getColumnLatestCell(hbaseColumn.getFirst(), hbaseColumn.getSecond());
- oneBlock.getCellBlock(colBlockIdx).set(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
- }
- return oneBlock;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void close() throws IOException {
- scanner.close();
- hbaseTable.close();
- }
- };
- }
-
- private Scan buildScan(GTRecord pkStart, GTRecord 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(makeRowKeyToScan(pkStart, (byte) 0x00));
- scan.setStopRow(makeRowKeyToScan(pkEnd, (byte) 0xff));
- //TODO fuzzy match
- return scan;
- }
-
- private byte[] makeRowKeyToScan(GTRecord pkRec, byte fill) {
- ByteArray pk = GTRecord.exportScanKey(pkRec);
- int pkMaxLen = info.getMaxColumnLength(info.getPrimaryKey());
-
- byte[] buf = new byte[pkMaxLen + RowConstants.ROWKEY_CUBOIDID_LEN];
- Arrays.fill(buf, fill);
-
- System.arraycopy(cuboid.getBytes(), 0, buf, 0, RowConstants.ROWKEY_CUBOIDID_LEN);
- if (pk != null && pk.array() != null) {
- System.arraycopy(pk.array(), pk.offset(), buf, RowConstants.ROWKEY_CUBOIDID_LEN, pk.length());
- }
- return buf;
- }
-
- private List<Pair<byte[], byte[]>> makeHBaseColumns(ImmutableBitSet selectedColBlocks) {
- List<Pair<byte[], byte[]>> result = Lists.newArrayList();
-
- int colBlockIdx = 1; // start from 1; the 0th column block is primary key which maps to rowkey
- 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[] byteQualifier = Bytes.toBytes(hbaseColDesc.getQualifier());
- result.add(new Pair<byte[], byte[]>(byteFamily, byteQualifier));
- }
- colBlockIdx++;
- }
- }
-
- return result;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
new file mode 100644
index 0000000..d1b7d4e
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
@@ -0,0 +1,78 @@
+package org.apache.kylin.storage.cube;
+
+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.common.util.Pair;
+import org.apache.kylin.cube.CubeSegment;
+import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRawScanner;
+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;
+
+/**
+ * for test use only
+ */
+public class CubeHBaseScanRPC extends CubeHBaseRPC {
+
+ 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());
+ final List<Pair<byte[], byte[]>> hbaseColumns = makeHBaseColumns(selectedColBlocks);
+
+ RawScan rawScan = prepareRawScan(scanRequest.getPkStart(), scanRequest.getPkEnd(), hbaseColumns);
+ Scan hbaseScan = CubeHBaseRPC.buildScan(rawScan);
+
+ final ResultScanner scanner = hbaseTable.getScanner(hbaseScan);
+ final Iterator<Result> iterator = scanner.iterator();
+
+ CellListIterator cellListIterator = new CellListIterator() {
+ @Override
+ public void close() throws IOException {
+ scanner.close();
+ hbaseTable.close();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public List<Cell> next() {
+ return iterator.next().listCells();
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+
+ IGTStore store = new HBaseGTStore(cellListIterator, scanRequest, hbaseColumns);
+ GTRawScanner rawScanner = new GTRawScanner(scanRequest.getInfo(), store, scanRequest);
+ return scanRequest.decorateScanner(rawScanner);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
index 6af83d1..2411a6c 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
@@ -3,29 +3,34 @@ package org.apache.kylin.storage.cube;
import java.io.IOException;
import java.util.BitSet;
import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Set;
import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.storage.util.KryoUtils;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.gridtable.CubeGridTable;
import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
+import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.gridtable.GTInfo;
-import org.apache.kylin.gridtable.GTRawScanner;
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRange;
import org.apache.kylin.gridtable.GTScanRangePlanner;
import org.apache.kylin.gridtable.GTScanRequest;
import org.apache.kylin.gridtable.GTUtil;
+import org.apache.kylin.gridtable.IGTCodeSystem;
import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
public class CubeScanner implements IGTScanner {
@@ -33,34 +38,59 @@ public class CubeScanner implements IGTScanner {
final CubeSegment cubeSeg;
final GTInfo info;
- final CubeHBaseReadonlyStore store;
+ final byte[] trimmedInfoBytes;
final List<GTScanRequest> scanRequests;
final Scanner scanner;
+ final Cuboid cuboid;
public CubeScanner(CubeSegment cubeSeg, Cuboid cuboid, Set<TblColRef> dimensions, Set<TblColRef> groups, //
- Collection<FunctionDesc> metrics, TupleFilter filter) {
+ Collection<FunctionDesc> metrics, TupleFilter filter, boolean allowPreAggregate) {
+ this.cuboid = cuboid;
this.cubeSeg = cubeSeg;
this.info = CubeGridTable.newGTInfo(cubeSeg, cuboid.getId());
- this.store = new CubeHBaseReadonlyStore(info, cubeSeg, cuboid);
- CuboidToGridTableMapping mapping = new CuboidToGridTableMapping(cuboid);
+ CuboidToGridTableMapping mapping = cuboid.getCuboidToGridTableMapping();
+
+ //replace the constant values in filter to dictionary codes
TupleFilter gtFilter = GTUtil.convertFilterColumnsAndConstants(filter, info, mapping.getCuboidDimensionsInGTOrder(), groups);
+
ImmutableBitSet gtDimensions = makeGridTableColumns(mapping, dimensions);
- ImmutableBitSet gtAggrGroups = makeGridTableColumns(mapping, groups);
+ ImmutableBitSet gtAggrGroups = makeGridTableColumns(mapping, replaceDerivedColumns(groups, cubeSeg.getCubeDesc()));
ImmutableBitSet gtAggrMetrics = makeGridTableColumns(mapping, metrics);
- String[] gtAggrFuncs = makeAggrFuncs(metrics);
+ String[] gtAggrFuncs = makeAggrFuncs(mapping, metrics);
+ //TODO: should remove this in endpoint scenario
GTScanRangePlanner scanRangePlanner = new GTScanRangePlanner(info);
List<GTScanRange> scanRanges = scanRangePlanner.planScanRanges(gtFilter, MAX_SCAN_RANGES);
scanRequests = Lists.newArrayListWithCapacity(scanRanges.size());
+
+ IGTCodeSystem origin = info.trimCodeSystemForCoprocessor();
+ trimmedInfoBytes = KryoUtils.serialize(info);
+ info.setIGTCodeSystem(origin);
+ GTInfo trimmedInfo = KryoUtils.deserialize(this.trimmedInfoBytes, GTInfo.class);
+
for (GTScanRange range : scanRanges) {
- scanRequests.add(new GTScanRequest(info, range, gtDimensions, gtAggrGroups, gtAggrMetrics, gtAggrFuncs, gtFilter));
+ scanRequests.add(new GTScanRequest(trimmedInfo, range, gtDimensions, gtAggrGroups, gtAggrMetrics, gtAggrFuncs, gtFilter, allowPreAggregate));
}
-
+
scanner = new Scanner();
}
+ private Set<TblColRef> replaceDerivedColumns(Set<TblColRef> input, CubeDesc cubeDesc) {
+ Set<TblColRef> ret = Sets.newHashSet();
+ for (TblColRef col : input) {
+ if (cubeDesc.isDerived(col)) {
+ for (TblColRef host : cubeDesc.getHostInfo(col).columns) {
+ ret.add(host);
+ }
+ } else {
+ ret.add(col);
+ }
+ }
+ return ret;
+ }
+
private ImmutableBitSet makeGridTableColumns(CuboidToGridTableMapping mapping, Set<TblColRef> dimensions) {
BitSet result = new BitSet();
for (TblColRef dim : dimensions) {
@@ -82,10 +112,23 @@ public class CubeScanner implements IGTScanner {
return new ImmutableBitSet(result);
}
- private String[] makeAggrFuncs(Collection<FunctionDesc> metrics) {
- String[] result = new String[metrics.size()];
+ private String[] makeAggrFuncs(final CuboidToGridTableMapping mapping, Collection<FunctionDesc> metrics) {
+
+ //metrics are represented in ImmutableBitSet, which loses order information
+ //sort the aggrFuns to align with metrics natural order
+ List<FunctionDesc> metricList = Lists.newArrayList(metrics);
+ Collections.sort(metricList, new Comparator<FunctionDesc>() {
+ @Override
+ public int compare(FunctionDesc o1, FunctionDesc o2) {
+ int a = mapping.getIndexOf(o1);
+ int b = mapping.getIndexOf(o2);
+ return a - b;
+ }
+ });
+
+ String[] result = new String[metricList.size()];
int i = 0;
- for (FunctionDesc metric : metrics) {
+ for (FunctionDesc metric : metricList) {
result[i++] = metric.getExpression();
}
return result;
@@ -116,6 +159,38 @@ public class CubeScanner implements IGTScanner {
return scanner.getScannedRowBlockCount();
}
+ static class RemoteGTRecordAdapter implements Iterable<GTRecord> {
+
+ private final GTInfo info;
+ private final Iterator<GTRecord> input;
+
+ public RemoteGTRecordAdapter(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() {
+
+ }
+ };
+ }
+ }
+
private class Scanner {
final IGTScanner[] inputScanners = new IGTScanner[scanRequests.size()];
int cur = 0;
@@ -129,25 +204,27 @@ public class CubeScanner implements IGTScanner {
public boolean hasNext() {
if (next != null)
return true;
-
+
if (curIterator == null) {
if (cur >= scanRequests.size())
return false;
try {
- inputScanners[cur] = new GTRawScanner(info, store, scanRequests.get(cur));
+ CubeHBaseRPC rpc = new CubeHBaseEndpointRPC(cubeSeg, cuboid, info);
+ inputScanners[cur] = rpc.getGTScanner(scanRequests.get(cur));
curIterator = inputScanners[cur].iterator();
+ //curIterator = new RemoteGTRecordAdapter(info, inputScanners[cur].iterator()).iterator();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
-
+
if (curIterator.hasNext() == false) {
curIterator = null;
cur++;
return hasNext();
}
-
+
next = curIterator.next();
return true;
}
@@ -186,7 +263,7 @@ public class CubeScanner implements IGTScanner {
for (int i = 0; i < inputScanners.length; i++) {
if (inputScanners[i] == null)
break;
-
+
result += inputScanners[i].getScannedRowCount();
}
return result;
@@ -197,12 +274,12 @@ public class CubeScanner implements IGTScanner {
for (int i = 0; i < inputScanners.length; i++) {
if (inputScanners[i] == null)
break;
-
+
result += inputScanners[i].getScannedRowBlockCount();
}
return result;
}
}
-
+
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/CubeStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeStorageQuery.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeStorageQuery.java
index 8164310..e9225f9 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeStorageQuery.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeStorageQuery.java
@@ -1,11 +1,8 @@
package org.apache.kylin.storage.cube;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Set;
-
+import com.google.common.collect.Lists;
+import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
@@ -32,9 +29,7 @@ import org.apache.kylin.storage.tuple.TupleInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
-import com.google.common.collect.Sets;
+import java.util.*;
public class CubeStorageQuery implements ICachableStorageQuery {
@@ -82,7 +77,7 @@ public class CubeStorageQuery implements ICachableStorageQuery {
Set<TblColRef> singleValuesD = findSingleValueColumns(filter);
boolean isExactAggregation = isExactAggregation(cuboid, groups, filterDimsD, singleValuesD, derivedPostAggregation);
context.setExactAggregation(isExactAggregation);
-
+
if (isExactAggregation) {
metrics = replaceHolisticCountDistinct(metrics);
}
@@ -92,17 +87,17 @@ public class CubeStorageQuery implements ICachableStorageQuery {
setThreshold(dimensionsD, metrics, context); // set cautious threshold to prevent out of memory
// TODO enable coprocessor
-// setCoprocessor(groupsCopD, valueDecoders, context); // enable coprocessor if beneficial
+ // setCoprocessor(groupsCopD, valueDecoders, context); // enable coprocessor if beneficial
setLimit(filter, context);
List<CubeScanner> scanners = Lists.newArrayList();
for (CubeSegment cubeSeg : cubeInstance.getSegments(SegmentStatusEnum.READY)) {
- scanners.add(new CubeScanner(cubeSeg, cuboid, dimensionsD, groupsD, metrics, filterD));
+ scanners.add(new CubeScanner(cubeSeg, cuboid, dimensionsD, groupsD, metrics, filterD, !isExactAggregation));
}
-
+
if (scanners.isEmpty())
return ITupleIterator.EMPTY_TUPLE_ITERATOR;
-
+
return new SequentialCubeTupleIterator(scanners, cuboid, dimensionsD, metrics, returnTupleInfo);
}
@@ -122,7 +117,7 @@ public class CubeStorageQuery implements ICachableStorageQuery {
dimensions.add(column);
}
}
-
+
private FunctionDesc findAggrFuncFromCubeDesc(FunctionDesc aggrFunc) {
for (MeasureDesc measure : cubeDesc.getMeasures()) {
if (measure.getFunction().equals(aggrFunc))
@@ -232,7 +227,7 @@ public class CubeStorageQuery implements ICachableStorageQuery {
result.add(metric);
continue;
}
-
+
FunctionDesc holisticVersion = null;
for (MeasureDesc measure : cubeDesc.getMeasures()) {
FunctionDesc measureFunc = measure.getFunction();
@@ -349,7 +344,7 @@ public class CubeStorageQuery implements ICachableStorageQuery {
context.enableLimit();
}
}
-
+
// ============================================================================
@Override
@@ -362,7 +357,6 @@ public class CubeStorageQuery implements ICachableStorageQuery {
return cubeInstance.getUuid();
}
-
@Override
public boolean isDynamic() {
return false;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/CubeTupleConverter.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeTupleConverter.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeTupleConverter.java
index 0657a69..aefff7a 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeTupleConverter.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeTupleConverter.java
@@ -18,26 +18,28 @@
package org.apache.kylin.storage.cube;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-
+import com.google.common.collect.Lists;
import org.apache.kylin.common.util.Array;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
import org.apache.kylin.cube.model.CubeDesc.DeriveInfo;
import org.apache.kylin.dict.lookup.LookupStringTable;
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.metadata.model.FunctionDesc;
-import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.storage.tuple.Tuple;
import org.apache.kylin.storage.tuple.TupleInfo;
-import com.google.common.collect.Lists;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+/**
+ * convert GTRecord to tuple
+ */
public class CubeTupleConverter {
final CubeSegment cubeSeg;
@@ -58,7 +60,7 @@ public class CubeTupleConverter {
this.derivedColFillers = Lists.newArrayList();
List<TblColRef> cuboidDims = cuboid.getColumns();
- List<MeasureDesc> cuboidMeasures = cuboid.getCube().getMeasures();
+ CuboidToGridTableMapping mapping = cuboid.getCuboidToGridTableMapping();
nSelectedDims = selectedDimensions.size();
gtColIdx = new int[selectedDimensions.size() + selectedMetrics.size()];
@@ -68,32 +70,26 @@ public class CubeTupleConverter {
int iii = 0;
// pre-calculate dimension index mapping to tuple
- for (int i = 0; i < cuboidDims.size(); i++) {
- TblColRef col = cuboidDims.get(i);
- if (selectedDimensions.contains(col)) {
- gtColIdx[iii] = i;
- tupleIdx[iii] = tupleInfo.hasColumn(col) ? tupleInfo.getColumnIndex(col) : -1;
- iii++;
- }
+ for (TblColRef dim : selectedDimensions) {
+ int i = mapping.getIndexOf(dim);
+ gtColIdx[iii] = i;
+ tupleIdx[iii] = tupleInfo.hasColumn(dim) ? tupleInfo.getColumnIndex(dim) : -1;
+ iii++;
}
- // pre-calculate metrics index mapping to tuple
- for (int i = 0; i < cuboidMeasures.size(); i++) {
- FunctionDesc aggrFunc = cuboidMeasures.get(i).getFunction();
- if (contains(selectedMetrics, aggrFunc)) {
- gtColIdx[iii] = cuboidDims.size() + i;
- // a rewrite metrics is identified by its rewrite field name
- if (aggrFunc.needRewrite()) {
- String rewriteFieldName = aggrFunc.getRewriteFieldName();
- tupleIdx[iii] = tupleInfo.hasField(rewriteFieldName) ? tupleInfo.getFieldIndex(rewriteFieldName) : -1;
- }
- // a non-rewrite metrics (like sum, or dimension playing as metrics) is like a dimension column
- else {
- TblColRef col = aggrFunc.getParameter().getColRefs().get(0);
- tupleIdx[iii] = tupleInfo.hasColumn(col) ? tupleInfo.getColumnIndex(col) : -1;
- }
- iii++;
+ for (FunctionDesc metric : selectedMetrics) {
+ int i = mapping.getIndexOf(metric);
+ gtColIdx[iii] = i;
+ if (metric.needRewrite()) {
+ String rewriteFieldName = metric.getRewriteFieldName();
+ tupleIdx[iii] = tupleInfo.hasField(rewriteFieldName) ? tupleInfo.getFieldIndex(rewriteFieldName) : -1;
}
+ // a non-rewrite metrics (like sum, or dimension playing as metrics) is like a dimension column
+ else {
+ TblColRef col = metric.getParameter().getColRefs().get(0);
+ tupleIdx[iii] = tupleInfo.hasColumn(col) ? tupleInfo.getColumnIndex(col) : -1;
+ }
+ iii++;
}
// prepare derived columns and filler
@@ -109,18 +105,6 @@ public class CubeTupleConverter {
}
}
- private boolean contains(Set<FunctionDesc> selectedMetrics, FunctionDesc aggrFunc) {
- if (aggrFunc.isCountDistinct()) {
- // need special attention for count distinct and its holistic version, they are equals() but not exactly the same
- for (FunctionDesc selected : selectedMetrics) {
- if (selected.equals(aggrFunc) && selected.isHolisticCountDistinct() == aggrFunc.isHolisticCountDistinct())
- return true;
- }
- return false;
- } else
- return selectedMetrics.contains(aggrFunc);
- }
-
public void translateResult(GTRecord record, Tuple tuple) {
record.getValues(gtColIdx, tmpValues);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java b/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java
new file mode 100644
index 0000000..6231e3b
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java
@@ -0,0 +1,113 @@
+/*
+ * 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.kylin.storage.cube;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRowBlock;
+import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.gridtable.IGTStore;
+
+public class HBaseGTStore implements IGTStore {
+
+ private CellListIterator cellListIterator;
+
+ private GTInfo info;
+ private List<Pair<byte[], byte[]>> hbaseColumns;
+ private ImmutableBitSet selectedColBlocks;
+
+ private GTRowBlock oneBlock;
+
+ public HBaseGTStore(CellListIterator cellListIterator, GTScanRequest gtScanRequest, List<Pair<byte[], byte[]>> hbaseColumns) {
+ this.cellListIterator = cellListIterator;
+
+ this.info = gtScanRequest.getInfo();
+ this.hbaseColumns = hbaseColumns;
+ this.selectedColBlocks = gtScanRequest.getSelectedColBlocks().set(0);
+
+ oneBlock = new GTRowBlock(info); // avoid object creation
+ oneBlock.setNumberOfRows(1); // row block is always disabled for cubes, row block contains only one record
+ }
+
+ @Override
+ public GTInfo getInfo() {
+ return info;
+ }
+
+ @Override
+ public IGTStoreWriter rebuild(int shard) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public IGTStoreWriter append(int shard, GTRowBlock.Writer fillLast) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public IGTStoreScanner scan(GTScanRequest scanRequest) throws IOException {
+ return new IGTStoreScanner() {
+ @Override
+ public boolean hasNext() {
+ return cellListIterator.hasNext();
+ }
+
+ @Override
+ public GTRowBlock next() {
+ List<Cell> oneRow = cellListIterator.next();
+ if (oneRow.size() < 1) {
+ throw new IllegalStateException("cell list's size less than 1");
+ }
+
+ // dimensions, set to primary key, also the 0th column block
+ Cell firstCell = oneRow.get(0);
+ byte[] rowkey = firstCell.getRowArray();
+ oneBlock.getPrimaryKey().set(rowkey, RowConstants.ROWKEY_CUBOIDID_LEN + firstCell.getRowOffset(), firstCell.getRowLength() - RowConstants.ROWKEY_CUBOIDID_LEN);
+ oneBlock.getCellBlock(0).set(rowkey, RowConstants.ROWKEY_CUBOIDID_LEN + firstCell.getRowOffset(), firstCell.getRowLength() - RowConstants.ROWKEY_CUBOIDID_LEN);
+
+ // metrics
+ int hbaseColIdx = 0;
+ for (int i = 1; i < selectedColBlocks.trueBitCount(); i++) {
+ int colBlockIdx = selectedColBlocks.trueBitAt(i);
+ Pair<byte[], byte[]> hbaseColumn = hbaseColumns.get(hbaseColIdx++);
+ Cell cell = CubeHBaseRPC.findCell(oneRow, hbaseColumn.getFirst(), hbaseColumn.getSecond());
+ oneBlock.getCellBlock(colBlockIdx).set(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ }
+ return oneBlock;
+
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws IOException {
+ cellListIterator.close();
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/cube/RawScan.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/RawScan.java b/storage/src/main/java/org/apache/kylin/storage/cube/RawScan.java
new file mode 100644
index 0000000..c42e89f
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/RawScan.java
@@ -0,0 +1,40 @@
+/*
+ * 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.kylin.storage.cube;
+
+import java.util.List;
+
+import org.apache.kylin.common.util.Pair;
+
+public class RawScan {
+
+ public byte[] startKey;
+ public byte[] endKey;
+ public List<Pair<byte[], byte[]>> hbaseColumns;
+ public List<Pair<byte[], byte[]>> fuzzyKey;
+
+ public RawScan(byte[] startKey, byte[] endKey, List<Pair<byte[], byte[]>> hbaseColumns, List<Pair<byte[], byte[]>> fuzzyKey) {
+
+ this.startKey = startKey;
+ this.endKey = endKey;
+ this.hbaseColumns = hbaseColumns;
+ this.fuzzyKey = fuzzyKey;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/hbase/CubeStorageQuery.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/CubeStorageQuery.java b/storage/src/main/java/org/apache/kylin/storage/hbase/CubeStorageQuery.java
index 4c3920b..459e72d 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/CubeStorageQuery.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/CubeStorageQuery.java
@@ -405,6 +405,7 @@ public class CubeStorageQuery implements ICachableStorageQuery {
logger.info("READY segs count: " + segs.size());
// build row key range for each cube segment
+ StringBuilder sb = new StringBuilder("hbasekeyrange trace: ");
for (CubeSegment cubeSeg : segs) {
// consider derived (lookup snapshot), filter on dimension may
@@ -420,11 +421,16 @@ public class CubeStorageQuery implements ICachableStorageQuery {
scanRanges.add(rowKeyRange);
}
+ sb.append(scanRanges.size() + "=>");
List<HBaseKeyRange> mergedRanges = mergeOverlapRanges(scanRanges);
+ sb.append(mergedRanges.size() + "=>");
mergedRanges = mergeTooManyRanges(mergedRanges);
+ sb.append(mergedRanges.size() + ", ");
result.addAll(mergedRanges);
}
+ logger.info(sb.toString());
+
logger.info("hbasekeyrange count: " + result.size());
dropUnhitSegments(result);
logger.info("hbasekeyrange count after dropping unhit :" + result.size());
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
index 563f346..f4fd65d 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
@@ -37,6 +37,7 @@ public class DictCodeSystem implements IFilterCodeSystem<String> {
return tupleValue.compareTo(constValue);
}
+ //TODO: should use ISO-8859-1 rather than UTF8
@Override
public void serialize(String value, ByteBuffer buffer) {
BytesUtil.writeUTFString(value, buffer);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java
new file mode 100644
index 0000000..c1a242d
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java
@@ -0,0 +1,198 @@
+/*
+ * 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.kylin.storage.hbase.coprocessor.endpoint;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.Coprocessor;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+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.protobuf.ResponseConverter;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.kylin.common.util.CompressionUtils;
+import org.apache.kylin.storage.util.KryoUtils;
+import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.gridtable.GTRawScanner;
+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.cube.CellListIterator;
+import org.apache.kylin.storage.cube.CubeHBaseRPC;
+import org.apache.kylin.storage.cube.HBaseGTStore;
+import org.apache.kylin.storage.cube.RawScan;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitResponse.Builder;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitResponse.Stats;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.Service;
+
+public class CubeVisitService extends CubeVisitProtos.CubeVisitService implements Coprocessor, CoprocessorService {
+
+ private static final Logger logger = LoggerFactory.getLogger(CubeVisitService.class);
+ //TODO limit memory footprint
+ private static final int MEMORY_LIMIT = 500 * 1024 * 1024;
+
+ private RegionCoprocessorEnvironment env;
+
+ private long serviceStartTime;
+
+ static class InnerScannerAsIterator implements CellListIterator {
+ private RegionScanner regionScanner;
+ private List<Cell> nextOne = Lists.newArrayList();
+ private List<Cell> ret = Lists.newArrayList();
+
+ private boolean hasMore;
+
+ public InnerScannerAsIterator(RegionScanner regionScanner) {
+ this.regionScanner = regionScanner;
+
+ try {
+ hasMore = regionScanner.nextRaw(nextOne);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return !nextOne.isEmpty();
+ }
+
+ @Override
+ public List<Cell> next() {
+
+ if (nextOne.size() < 1) {
+ throw new IllegalStateException();
+ }
+ ret.clear();
+ ret.addAll(nextOne);
+ nextOne.clear();
+ try {
+ if (hasMore) {
+ hasMore = regionScanner.nextRaw(nextOne);
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return ret;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws IOException {
+ //does not need to close as regionScanner will be closed in finally block
+ }
+ }
+
+ @Override
+ public void visitCube(RpcController controller, CubeVisitProtos.CubeVisitRequest request, RpcCallback<CubeVisitProtos.CubeVisitResponse> done) {
+ RegionScanner innerScanner = null;
+ HRegion region = null;
+
+ try {
+ this.serviceStartTime = System.currentTimeMillis();
+
+ GTScanRequest scanReq = KryoUtils.deserialize(request.getGtScanRequest().toByteArray(), GTScanRequest.class);
+ RawScan hbaseRawScan = KryoUtils.deserialize(request.getHbaseRawScan().toByteArray(), RawScan.class);
+ Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan);
+
+ region = env.getRegion();
+ region.startRegionOperation();
+
+ innerScanner = region.getScanner(scan);
+ InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner);
+
+ IGTStore store = new HBaseGTStore(cellListIterator, scanReq, hbaseRawScan.hbaseColumns);
+ GTRawScanner rawScanner = new GTRawScanner(scanReq.getInfo(), store, scanReq);
+ IGTScanner finalScanner = scanReq.decorateScanner(rawScanner);
+
+ ByteBuffer buffer = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream(RowConstants.ROWVALUE_BUFFER_SIZE);
+ for (GTRecord oneRecord : finalScanner) {
+ buffer.clear();
+ oneRecord.exportAllColumns(buffer);
+ buffer.flip();
+ outputStream.write(buffer.array(), buffer.arrayOffset() - buffer.position(), buffer.remaining());
+ }
+ //outputStream.close() is not necessary
+ byte[] allRows = outputStream.toByteArray();
+ Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder();
+ done.run(responseBuilder.//
+ setCompressedRows(ByteString.copyFrom(CompressionUtils.compress(allRows))).//too many array copies
+ setStats(Stats.newBuilder().//
+ setAggregatedRowCount(0).//
+ setScannedRowCount(0).//
+ setServiceStartTime(serviceStartTime).//
+ setServiceEndTime(System.currentTimeMillis()).build()).//
+ build());
+
+ } catch (IOException ioe) {
+ logger.error(ioe.toString());
+ ResponseConverter.setControllerException(controller, ioe);
+ } finally {
+ IOUtils.closeQuietly(innerScanner);
+ if (region != null) {
+ try {
+ region.closeRegionOperation();
+ } catch (IOException e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+
+ @Override
+ public void start(CoprocessorEnvironment env) throws IOException {
+ if (env instanceof RegionCoprocessorEnvironment) {
+ this.env = (RegionCoprocessorEnvironment) env;
+ } else {
+ throw new CoprocessorException("Must be loaded on a table region!");
+ }
+ }
+
+ @Override
+ public void stop(CoprocessorEnvironment env) throws IOException {
+ }
+
+ @Override
+ public Service getService() {
+ return this;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/ae627c3b/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 429e060..5010680 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
@@ -79,6 +79,7 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
private Scan prepareScan(IIProtos.IIRequest request, HRegion region) throws IOException {
Scan scan = new Scan();
+
scan.addColumn(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_QUALIFIER_BYTES);
scan.addColumn(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_DICTIONARY_BYTES);