You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by sh...@apache.org on 2015/08/26 16:08:49 UTC
[26/34] incubator-kylin git commit: KYLIN-960 organize existing
storage module
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeTupleConverter.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeTupleConverter.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeTupleConverter.java
new file mode 100644
index 0000000..fb54cce
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/CubeTupleConverter.java
@@ -0,0 +1,224 @@
+/*
+ * 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.cube.v2;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+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.TblColRef;
+import org.apache.kylin.storage.tuple.Tuple;
+import org.apache.kylin.storage.tuple.TupleInfo;
+
+import com.google.common.collect.Lists;
+
+/**
+ * convert GTRecord to tuple
+ */
+public class CubeTupleConverter {
+
+ final CubeSegment cubeSeg;
+ final Cuboid cuboid;
+ final TupleInfo tupleInfo;
+ final List<IDerivedColumnFiller> derivedColFillers;
+
+ final int[] gtColIdx;
+ final int[] tupleIdx;
+ final Object[] tmpValues;
+ final int nSelectedDims;
+
+ public CubeTupleConverter(CubeSegment cubeSeg, Cuboid cuboid, //
+ Set<TblColRef> selectedDimensions, Set<FunctionDesc> selectedMetrics, TupleInfo returnTupleInfo) {
+ this.cubeSeg = cubeSeg;
+ this.cuboid = cuboid;
+ this.tupleInfo = returnTupleInfo;
+ this.derivedColFillers = Lists.newArrayList();
+
+ List<TblColRef> cuboidDims = cuboid.getColumns();
+ CuboidToGridTableMapping mapping = cuboid.getCuboidToGridTableMapping();
+
+ nSelectedDims = selectedDimensions.size();
+ gtColIdx = new int[selectedDimensions.size() + selectedMetrics.size()];
+ tupleIdx = new int[selectedDimensions.size() + selectedMetrics.size()];
+ tmpValues = new Object[selectedDimensions.size() + selectedMetrics.size()];
+
+ int iii = 0;
+
+ // pre-calculate dimension index mapping to tuple
+ for (TblColRef dim : selectedDimensions) {
+ int i = mapping.getIndexOf(dim);
+ gtColIdx[iii] = i;
+ tupleIdx[iii] = tupleInfo.hasColumn(dim) ? tupleInfo.getColumnIndex(dim) : -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
+ Map<Array<TblColRef>, List<DeriveInfo>> hostToDerivedInfo = cuboid.getCube().getHostToDerivedInfo(cuboidDims, null);
+ for (Entry<Array<TblColRef>, List<DeriveInfo>> entry : hostToDerivedInfo.entrySet()) {
+ TblColRef[] hostCols = entry.getKey().data;
+ for (DeriveInfo deriveInfo : entry.getValue()) {
+ IDerivedColumnFiller filler = newDerivedColumnFiller(hostCols, deriveInfo);
+ if (filler != null) {
+ derivedColFillers.add(filler);
+ }
+ }
+ }
+ }
+
+ public void translateResult(GTRecord record, Tuple tuple) {
+
+ record.getValues(gtColIdx, tmpValues);
+
+ // dimensions
+ for (int i = 0; i < nSelectedDims; i++) {
+ int ti = tupleIdx[i];
+ if (ti >= 0) {
+ tuple.setDimensionValue(ti, toString(tmpValues[i]));
+ }
+ }
+
+ // measures
+ for (int i = nSelectedDims; i < gtColIdx.length; i++) {
+ int ti = tupleIdx[i];
+ if (ti >= 0) {
+ tuple.setMeasureValue(ti, tmpValues[i]);
+ }
+ }
+
+ // derived
+ for (IDerivedColumnFiller filler : derivedColFillers) {
+ filler.fillDerivedColumns(tmpValues, tuple);
+ }
+ }
+
+ private interface IDerivedColumnFiller {
+ public void fillDerivedColumns(Object[] tmpValues, Tuple tuple);
+ }
+
+ private IDerivedColumnFiller newDerivedColumnFiller(TblColRef[] hostCols, final DeriveInfo deriveInfo) {
+ boolean allHostsPresent = true;
+ final int[] hostTmpIdx = new int[hostCols.length];
+ for (int i = 0; i < hostCols.length; i++) {
+ hostTmpIdx[i] = indexOnTheTmpValues(hostCols[i]);
+ allHostsPresent = allHostsPresent && hostTmpIdx[i] >= 0;
+ }
+
+ boolean needCopyDerived = false;
+ final int[] derivedTupleIdx = new int[deriveInfo.columns.length];
+ for (int i = 0; i < deriveInfo.columns.length; i++) {
+ TblColRef col = deriveInfo.columns[i];
+ derivedTupleIdx[i] = tupleInfo.hasColumn(col) ? tupleInfo.getColumnIndex(col) : -1;
+ needCopyDerived = needCopyDerived || derivedTupleIdx[i] >= 0;
+ }
+
+ if ((allHostsPresent && needCopyDerived) == false)
+ return null;
+
+ switch (deriveInfo.type) {
+ case LOOKUP:
+ return new IDerivedColumnFiller() {
+ CubeManager cubeMgr = CubeManager.getInstance(cubeSeg.getCubeInstance().getConfig());
+ LookupStringTable lookupTable = cubeMgr.getLookupTable(cubeSeg, deriveInfo.dimension);
+ int[] derivedColIdx = initDerivedColIdx();
+ Array<String> lookupKey = new Array<String>(new String[hostTmpIdx.length]);
+
+ private int[] initDerivedColIdx() {
+ int[] idx = new int[deriveInfo.columns.length];
+ for (int i = 0; i < idx.length; i++) {
+ idx[i] = deriveInfo.columns[i].getColumnDesc().getZeroBasedIndex();
+ }
+ return idx;
+ }
+
+ @Override
+ public void fillDerivedColumns(Object[] tmpValues, Tuple tuple) {
+ for (int i = 0; i < hostTmpIdx.length; i++) {
+ lookupKey.data[i] = CubeTupleConverter.toString(tmpValues[hostTmpIdx[i]]);
+ }
+
+ String[] lookupRow = lookupTable.getRow(lookupKey);
+
+ if (lookupRow != null) {
+ for (int i = 0; i < derivedTupleIdx.length; i++) {
+ if (derivedTupleIdx[i] >= 0) {
+ String value = lookupRow[derivedColIdx[i]];
+ tuple.setDimensionValue(derivedTupleIdx[i], value);
+ }
+ }
+ } else {
+ for (int i = 0; i < derivedTupleIdx.length; i++) {
+ if (derivedTupleIdx[i] >= 0) {
+ tuple.setDimensionValue(derivedTupleIdx[i], null);
+ }
+ }
+ }
+ }
+ };
+ case PK_FK:
+ return new IDerivedColumnFiller() {
+ @Override
+ public void fillDerivedColumns(Object[] tmpValues, Tuple tuple) {
+ // composite keys are split, so only copy [0] is enough, see CubeDesc.initDimensionColumns()
+ tuple.setDimensionValue(derivedTupleIdx[0], CubeTupleConverter.toString(tmpValues[hostTmpIdx[0]]));
+ }
+ };
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+
+ private int indexOnTheTmpValues(TblColRef col) {
+ List<TblColRef> cuboidDims = cuboid.getColumns();
+ int cuboidIdx = cuboidDims.indexOf(col);
+ for (int i = 0; i < gtColIdx.length; i++) {
+ if (gtColIdx[i] == cuboidIdx)
+ return i;
+ }
+ return -1;
+ }
+
+ private static String toString(Object o) {
+ return o == null ? null : o.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java
new file mode 100644
index 0000000..a6c6a23
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/HBaseReadonlyStore.java
@@ -0,0 +1,138 @@
+/*
+ * 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.cube.v2;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+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.GTRecord;
+import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.gridtable.IGTScanner;
+import org.apache.kylin.gridtable.IGTStore;
+import org.apache.kylin.gridtable.IGTWriter;
+
+public class HBaseReadonlyStore implements IGTStore {
+
+ private CellListIterator cellListIterator;
+
+ private GTInfo info;
+ private List<Pair<byte[], byte[]>> hbaseColumns;
+ private ImmutableBitSet selectedColBlocks;
+
+ public HBaseReadonlyStore(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);
+ }
+
+ @Override
+ public GTInfo getInfo() {
+ return info;
+ }
+
+ @Override
+ public IGTWriter rebuild(int shard) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public IGTWriter append(int shard) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public IGTScanner scan(GTScanRequest scanRequest) throws IOException {
+ return new IGTScanner() {
+ int count;
+
+ @Override
+ public void close() throws IOException {
+ cellListIterator.close();
+ }
+
+ @Override
+ public Iterator<GTRecord> iterator() {
+ return new Iterator<GTRecord>() {
+ GTRecord oneRecord = new GTRecord(info); // avoid object creation
+
+ @Override
+ public boolean hasNext() {
+ return cellListIterator.hasNext();
+ }
+
+ @Override
+ public GTRecord next() {
+ List<Cell> oneRow = cellListIterator.next();
+ if (oneRow.size() < 1) {
+ throw new IllegalStateException("cell list's size less than 1");
+ }
+
+ ByteBuffer buf;
+
+ // dimensions, set to primary key, also the 0th column block
+ Cell firstCell = oneRow.get(0);
+ buf = byteBuffer(firstCell.getRowArray(), RowConstants.ROWKEY_CUBOIDID_LEN + firstCell.getRowOffset(), firstCell.getRowLength() - RowConstants.ROWKEY_CUBOIDID_LEN);
+ oneRecord.loadCellBlock(0, buf);
+
+ // 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());
+ buf = byteBuffer(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ oneRecord.loadCellBlock(colBlockIdx, buf);
+ }
+ return oneRecord;
+
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ private ByteBuffer byteBuffer(byte[] array, int offset, int length) {
+ return ByteBuffer.wrap(array, offset, length);
+ }
+
+ };
+ }
+
+ @Override
+ public GTInfo getInfo() {
+ return info;
+ }
+
+ @Override
+ public int getScannedRowCount() {
+ return count;
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/RawScan.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/RawScan.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/RawScan.java
new file mode 100644
index 0000000..aa73927
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/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.hbase.cube.v2;
+
+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/11611cc2/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
new file mode 100644
index 0000000..432fc04
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/SequentialCubeTupleIterator.java
@@ -0,0 +1,111 @@
+package org.apache.kylin.storage.hbase.cube.v2;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+
+import org.apache.kylin.cube.cuboid.Cuboid;
+import org.apache.kylin.gridtable.GTRecord;
+import org.apache.kylin.metadata.model.FunctionDesc;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.tuple.ITuple;
+import org.apache.kylin.metadata.tuple.ITupleIterator;
+import org.apache.kylin.storage.tuple.Tuple;
+import org.apache.kylin.storage.tuple.TupleInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SequentialCubeTupleIterator implements ITupleIterator {
+
+ private static final Logger logger = LoggerFactory.getLogger(SequentialCubeTupleIterator.class);
+
+ private final Cuboid cuboid;
+ private final Set<TblColRef> selectedDimensions;
+ private final Set<FunctionDesc> selectedMetrics;
+ private final TupleInfo tupleInfo;
+ private final Tuple tuple;
+ private final Iterator<CubeScanner> scannerIterator;
+
+ private CubeScanner curScanner;
+ private Iterator<GTRecord> curRecordIterator;
+ private CubeTupleConverter curTupleConverter;
+ private Tuple next;
+
+ public SequentialCubeTupleIterator(List<CubeScanner> scanners, Cuboid cuboid, Set<TblColRef> selectedDimensions, //
+ Set<FunctionDesc> selectedMetrics, TupleInfo returnTupleInfo) {
+ this.cuboid = cuboid;
+ this.selectedDimensions = selectedDimensions;
+ this.selectedMetrics = selectedMetrics;
+ this.tupleInfo = returnTupleInfo;
+ this.tuple = new Tuple(returnTupleInfo);
+ this.scannerIterator = scanners.iterator();
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (next != null)
+ return true;
+
+ if (curScanner == null) {
+ if (scannerIterator.hasNext()) {
+ curScanner = scannerIterator.next();
+ curRecordIterator = curScanner.iterator();
+ curTupleConverter = new CubeTupleConverter(curScanner.cubeSeg, cuboid, selectedDimensions, selectedMetrics, tupleInfo);
+ } else {
+ return false;
+ }
+ }
+
+ if (curRecordIterator.hasNext()) {
+ curTupleConverter.translateResult(curRecordIterator.next(), tuple);
+ next = tuple;
+ return true;
+ } else {
+ close(curScanner);
+ curScanner = null;
+ curRecordIterator = null;
+ curTupleConverter = null;
+ return hasNext();
+ }
+ }
+
+ @Override
+ public ITuple next() {
+ // fetch next record
+ if (next == null) {
+ hasNext();
+ if (next == null)
+ throw new NoSuchElementException();
+ }
+
+ ITuple result = next;
+ next = null;
+ return result;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() {
+ if (curScanner != null)
+ close(curScanner);
+
+ while (scannerIterator.hasNext()) {
+ close(scannerIterator.next());
+ }
+ }
+
+ private void close(CubeScanner scanner) {
+ try {
+ scanner.close();
+ } catch (IOException e) {
+ logger.error("Exception when close CubeScanner", e);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
new file mode 100644
index 0000000..f0b8c6f
--- /dev/null
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
@@ -0,0 +1,196 @@
+/*
+ * 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.cube.v2.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.cube.kv.RowConstants;
+import org.apache.kylin.cube.util.KryoUtils;
+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.cube.v2.CellListIterator;
+import org.apache.kylin.storage.hbase.cube.v2.CubeHBaseRPC;
+import org.apache.kylin.storage.hbase.cube.v2.HBaseReadonlyStore;
+import org.apache.kylin.storage.hbase.cube.v2.RawScan;
+import org.apache.kylin.storage.hbase.cube.v2.coprocessor.endpoint.generated.CubeVisitProtos;
+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);
+ //TODO: rewrite own start/end
+ Scan scan = CubeHBaseRPC.buildScan(hbaseRawScan);
+
+ region = env.getRegion();
+ region.startRegionOperation();
+
+ innerScanner = region.getScanner(scan);
+ InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner);
+
+ IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScan.hbaseColumns);
+ IGTScanner rawScanner = store.scan(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();
+ CubeVisitProtos.CubeVisitResponse.Builder responseBuilder = CubeVisitProtos.CubeVisitResponse.newBuilder();
+ done.run(responseBuilder.//
+ setCompressedRows(ByteString.copyFrom(CompressionUtils.compress(allRows))).//too many array copies
+ setStats(CubeVisitProtos.CubeVisitResponse.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;
+ }
+}