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:41 UTC
[18/34] incubator-kylin git commit: KYLIN-960 organize existing
storage module
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/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
deleted file mode 100644
index d196d63..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
+++ /dev/null
@@ -1,160 +0,0 @@
-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/11611cc2/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
deleted file mode 100644
index ce0ef55..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
+++ /dev/null
@@ -1,77 +0,0 @@
-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.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 HBaseReadonlyStore(cellListIterator, scanRequest, hbaseColumns);
- IGTScanner rawScanner = store.scan(scanRequest);
- return scanRequest.decorateScanner(rawScanner);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/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
deleted file mode 100644
index 8368902..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
+++ /dev/null
@@ -1,265 +0,0 @@
-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.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.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.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 {
-
- private static final int MAX_SCAN_RANGES = 200;
-
- final CubeSegment cubeSeg;
- final GTInfo info;
- 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, boolean allowPreAggregate) {
- this.cuboid = cuboid;
- this.cubeSeg = cubeSeg;
- this.info = CubeGridTable.newGTInfo(cubeSeg, cuboid.getId());
-
- 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, replaceDerivedColumns(groups, cubeSeg.getCubeDesc()));
- ImmutableBitSet gtAggrMetrics = makeGridTableColumns(mapping, 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());
-
- trimmedInfoBytes = GTInfo.serialize(info);
- GTInfo trimmedInfo = GTInfo.deserialize(trimmedInfoBytes);
-
- for (GTScanRange range : scanRanges) {
- 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) {
- int idx = mapping.getIndexOf(dim);
- if (idx >= 0)
- result.set(idx);
- }
- return new ImmutableBitSet(result);
- }
-
- private ImmutableBitSet makeGridTableColumns(CuboidToGridTableMapping mapping, Collection<FunctionDesc> metrics) {
- BitSet result = new BitSet();
- for (FunctionDesc metric : metrics) {
- int idx = mapping.getIndexOf(metric);
- if (idx < 0)
- throw new IllegalStateException(metric + " not found in " + mapping);
- result.set(idx);
- }
- return new ImmutableBitSet(result);
- }
-
- 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 : metricList) {
- result[i++] = metric.getExpression();
- }
- return result;
- }
-
- @Override
- public Iterator<GTRecord> iterator() {
- return scanner.iterator();
- }
-
- @Override
- public void close() throws IOException {
- scanner.close();
- }
-
- @Override
- public GTInfo getInfo() {
- return info;
- }
-
- @Override
- public int getScannedRowCount() {
- return scanner.getScannedRowCount();
- }
-
- 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;
- Iterator<GTRecord> curIterator = null;
- GTRecord next = null;
-
- public Iterator<GTRecord> iterator() {
- return new Iterator<GTRecord>() {
-
- @Override
- public boolean hasNext() {
- if (next != null)
- return true;
-
- if (curIterator == null) {
- if (cur >= scanRequests.size())
- return false;
-
- try {
- 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;
- }
-
- @Override
- public GTRecord next() {
- // fetch next record
- if (next == null) {
- hasNext();
- if (next == null)
- throw new NoSuchElementException();
- }
-
- GTRecord result = next;
- next = null;
- return result;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
- };
- }
-
- public void close() throws IOException {
- for (int i = 0; i < inputScanners.length; i++) {
- if (inputScanners[i] != null) {
- inputScanners[i].close();
- }
- }
- }
-
- public int getScannedRowCount() {
- int result = 0;
- for (int i = 0; i < inputScanners.length; i++) {
- if (inputScanners[i] == null)
- break;
-
- result += inputScanners[i].getScannedRowCount();
- }
- return result;
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/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
deleted file mode 100644
index 7427fc5..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeStorageQuery.java
+++ /dev/null
@@ -1,370 +0,0 @@
-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 org.apache.kylin.common.util.Pair;
-import org.apache.kylin.cube.CubeInstance;
-import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.cube.model.CubeDesc.DeriveInfo;
-import org.apache.kylin.dict.lookup.LookupStringTable;
-import org.apache.kylin.metadata.filter.ColumnTupleFilter;
-import org.apache.kylin.metadata.filter.CompareTupleFilter;
-import org.apache.kylin.metadata.filter.LogicalTupleFilter;
-import org.apache.kylin.metadata.filter.TupleFilter;
-import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
-import org.apache.kylin.metadata.model.FunctionDesc;
-import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.SegmentStatusEnum;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.storage.ICachableStorageQuery;
-import org.apache.kylin.storage.StorageContext;
-import org.apache.kylin.storage.hbase.DerivedFilterTranslator;
-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;
-
-public class CubeStorageQuery implements ICachableStorageQuery {
-
- private static final Logger logger = LoggerFactory.getLogger(CubeStorageQuery.class);
-
- private static final long MEM_BUDGET_PER_QUERY = 3L * 1024 * 1024 * 1024; // 3G
-
- private final CubeInstance cubeInstance;
- private final CubeDesc cubeDesc;
-
- public CubeStorageQuery(CubeInstance cube) {
- this.cubeInstance = cube;
- this.cubeDesc = cube.getDescriptor();
- }
-
- @Override
- public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
- Collection<TblColRef> groups = sqlDigest.groupbyColumns;
- TupleFilter filter = sqlDigest.filter;
-
- // build dimension & metrics
- Set<TblColRef> dimensions = new LinkedHashSet<TblColRef>();
- Set<FunctionDesc> metrics = new LinkedHashSet<FunctionDesc>();
- buildDimensionsAndMetrics(sqlDigest, dimensions, metrics);
-
- // all dimensions = groups + filter dimensions
- Set<TblColRef> filterDims = Sets.newHashSet(dimensions);
- filterDims.removeAll(groups);
-
- // expand derived (xxxD means contains host columns only, derived columns were translated)
- Set<TblColRef> derivedPostAggregation = Sets.newHashSet();
- Set<TblColRef> groupsD = expandDerived(groups, derivedPostAggregation);
- Set<TblColRef> filterDimsD = expandDerived(filterDims, derivedPostAggregation);
- filterDimsD.removeAll(groupsD);
- derivedPostAggregation.removeAll(groups);
-
- // identify cuboid
- Set<TblColRef> dimensionsD = new LinkedHashSet<TblColRef>();
- dimensionsD.addAll(groupsD);
- dimensionsD.addAll(filterDimsD);
- Cuboid cuboid = identifyCuboid(dimensionsD);
- context.setCuboid(cuboid);
-
- // isExactAggregation? meaning: tuples returned from storage requires no further aggregation in query engine
- Set<TblColRef> singleValuesD = findSingleValueColumns(filter);
- boolean isExactAggregation = isExactAggregation(cuboid, groups, filterDimsD, singleValuesD, derivedPostAggregation);
- context.setExactAggregation(isExactAggregation);
-
- if (isExactAggregation) {
- metrics = replaceHolisticCountDistinct(metrics);
- }
-
- // replace derived columns in filter with host columns; columns on loosened condition must be added to group by
- TupleFilter filterD = translateDerived(filter, groupsD);
-
- setThreshold(dimensionsD, metrics, context); // set cautious threshold to prevent out of memory
- // TODO enable coprocessor
- // 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, !isExactAggregation));
- }
-
- if (scanners.isEmpty())
- return ITupleIterator.EMPTY_TUPLE_ITERATOR;
-
- return new SequentialCubeTupleIterator(scanners, cuboid, dimensionsD, metrics, returnTupleInfo);
- }
-
- private void buildDimensionsAndMetrics(SQLDigest sqlDigest, Collection<TblColRef> dimensions, Collection<FunctionDesc> metrics) {
- for (FunctionDesc func : sqlDigest.aggregations) {
- if (!func.isDimensionAsMetric()) {
- // use the FunctionDesc from cube desc as much as possible, that has more info such as HLLC precision
- metrics.add(findAggrFuncFromCubeDesc(func));
- }
- }
-
- for (TblColRef column : sqlDigest.allColumns) {
- // skip measure columns
- if (sqlDigest.metricColumns.contains(column)) {
- continue;
- }
- dimensions.add(column);
- }
- }
-
- private FunctionDesc findAggrFuncFromCubeDesc(FunctionDesc aggrFunc) {
- for (MeasureDesc measure : cubeDesc.getMeasures()) {
- if (measure.getFunction().equals(aggrFunc))
- return measure.getFunction();
- }
- return aggrFunc;
- }
-
- private Set<TblColRef> expandDerived(Collection<TblColRef> cols, Set<TblColRef> derivedPostAggregation) {
- Set<TblColRef> expanded = Sets.newHashSet();
- for (TblColRef col : cols) {
- if (cubeDesc.isDerived(col)) {
- DeriveInfo hostInfo = cubeDesc.getHostInfo(col);
- for (TblColRef hostCol : hostInfo.columns) {
- expanded.add(hostCol);
- if (hostInfo.isOneToOne == false)
- derivedPostAggregation.add(hostCol);
- }
- } else {
- expanded.add(col);
- }
- }
- return expanded;
- }
-
- private Cuboid identifyCuboid(Set<TblColRef> dimensions) {
- long cuboidID = 0;
- for (TblColRef column : dimensions) {
- int index = cubeDesc.getRowkey().getColumnBitIndex(column);
- cuboidID |= 1L << index;
- }
- return Cuboid.findById(cubeDesc, cuboidID);
- }
-
- @SuppressWarnings("unchecked")
- private Set<TblColRef> findSingleValueColumns(TupleFilter filter) {
- Collection<? extends TupleFilter> toCheck;
- if (filter instanceof CompareTupleFilter) {
- toCheck = Collections.singleton(filter);
- } else if (filter instanceof LogicalTupleFilter && filter.getOperator() == FilterOperatorEnum.AND) {
- toCheck = filter.getChildren();
- } else {
- return (Set<TblColRef>) Collections.EMPTY_SET;
- }
-
- Set<TblColRef> result = Sets.newHashSet();
- for (TupleFilter f : toCheck) {
- if (f instanceof CompareTupleFilter) {
- CompareTupleFilter compFilter = (CompareTupleFilter) f;
- // is COL=const ?
- if (compFilter.getOperator() == FilterOperatorEnum.EQ && compFilter.getValues().size() == 1 && compFilter.getColumn() != null) {
- result.add(compFilter.getColumn());
- }
- }
- }
-
- // expand derived
- Set<TblColRef> resultD = Sets.newHashSet();
- for (TblColRef col : result) {
- if (cubeDesc.isDerived(col)) {
- DeriveInfo hostInfo = cubeDesc.getHostInfo(col);
- if (hostInfo.isOneToOne) {
- for (TblColRef hostCol : hostInfo.columns) {
- resultD.add(hostCol);
- }
- }
- //if not one2one, it will be pruned
- } else {
- resultD.add(col);
- }
- }
- return resultD;
- }
-
- private boolean isExactAggregation(Cuboid cuboid, Collection<TblColRef> groups, Set<TblColRef> othersD, Set<TblColRef> singleValuesD, Set<TblColRef> derivedPostAggregation) {
- boolean exact = true;
-
- if (cuboid.requirePostAggregation()) {
- exact = false;
- logger.info("exactAggregation is false because cuboid " + cuboid.getInputID() + "=> " + cuboid.getId());
- }
-
- // derived aggregation is bad, unless expanded columns are already in group by
- if (groups.containsAll(derivedPostAggregation) == false) {
- exact = false;
- logger.info("exactAggregation is false because derived column require post aggregation: " + derivedPostAggregation);
- }
-
- // other columns (from filter) is bad, unless they are ensured to have single value
- if (singleValuesD.containsAll(othersD) == false) {
- exact = false;
- logger.info("exactAggregation is false because some column not on group by: " + othersD //
- + " (single value column: " + singleValuesD + ")");
- }
-
- if (exact) {
- logger.info("exactAggregation is true");
- }
- return exact;
- }
-
- private Set<FunctionDesc> replaceHolisticCountDistinct(Set<FunctionDesc> metrics) {
- // for count distinct, try use its holistic version if possible
- Set<FunctionDesc> result = new LinkedHashSet<FunctionDesc>();
- for (FunctionDesc metric : metrics) {
- if (metric.isCountDistinct() == false) {
- result.add(metric);
- continue;
- }
-
- FunctionDesc holisticVersion = null;
- for (MeasureDesc measure : cubeDesc.getMeasures()) {
- FunctionDesc measureFunc = measure.getFunction();
- if (measureFunc.equals(metric) && measureFunc.isHolisticCountDistinct()) {
- holisticVersion = measureFunc;
- }
- }
- result.add(holisticVersion == null ? metric : holisticVersion);
- }
- return result;
- }
-
- @SuppressWarnings("unchecked")
- private TupleFilter translateDerived(TupleFilter filter, Set<TblColRef> collector) {
- if (filter == null)
- return filter;
-
- if (filter instanceof CompareTupleFilter) {
- return translateDerivedInCompare((CompareTupleFilter) filter, collector);
- }
-
- List<TupleFilter> children = (List<TupleFilter>) filter.getChildren();
- List<TupleFilter> newChildren = Lists.newArrayListWithCapacity(children.size());
- boolean modified = false;
- for (TupleFilter child : children) {
- TupleFilter translated = translateDerived(child, collector);
- newChildren.add(translated);
- if (child != translated)
- modified = true;
- }
- if (modified) {
- filter = replaceChildren(filter, newChildren);
- }
- return filter;
- }
-
- private TupleFilter replaceChildren(TupleFilter filter, List<TupleFilter> newChildren) {
- if (filter instanceof LogicalTupleFilter) {
- LogicalTupleFilter r = new LogicalTupleFilter(filter.getOperator());
- r.addChildren(newChildren);
- return r;
- } else
- throw new IllegalStateException("Cannot replaceChildren on " + filter);
- }
-
- private TupleFilter translateDerivedInCompare(CompareTupleFilter compf, Set<TblColRef> collector) {
- if (compf.getColumn() == null || compf.getValues().isEmpty())
- return compf;
-
- TblColRef derived = compf.getColumn();
- if (cubeDesc.isDerived(derived) == false)
- return compf;
-
- DeriveInfo hostInfo = cubeDesc.getHostInfo(derived);
- CubeManager cubeMgr = CubeManager.getInstance(this.cubeInstance.getConfig());
- CubeSegment seg = cubeInstance.getLatestReadySegment();
- LookupStringTable lookup = cubeMgr.getLookupTable(seg, hostInfo.dimension);
- Pair<TupleFilter, Boolean> translated = DerivedFilterTranslator.translate(lookup, hostInfo, compf);
- TupleFilter translatedFilter = translated.getFirst();
- boolean loosened = translated.getSecond();
- if (loosened) {
- collectColumnsRecursively(translatedFilter, collector);
- }
- return translatedFilter;
- }
-
- private void collectColumnsRecursively(TupleFilter filter, Set<TblColRef> collector) {
- if (filter instanceof ColumnTupleFilter) {
- collectColumns(((ColumnTupleFilter) filter).getColumn(), collector);
- }
- for (TupleFilter child : filter.getChildren()) {
- collectColumnsRecursively(child, collector);
- }
- }
-
- private void collectColumns(TblColRef col, Set<TblColRef> collector) {
- if (cubeDesc.isDerived(col)) {
- DeriveInfo hostInfo = cubeDesc.getHostInfo(col);
- for (TblColRef h : hostInfo.columns)
- collector.add(h);
- } else {
- collector.add(col);
- }
- }
-
- private void setThreshold(Collection<TblColRef> dimensions, Collection<FunctionDesc> metrics, StorageContext context) {
- boolean hasMemHungryCountDistinct = false;
- for (FunctionDesc func : metrics) {
- if (func.isCountDistinct() && !func.isHolisticCountDistinct()) {
- hasMemHungryCountDistinct = true;
- }
- }
-
- // need to limit the memory usage for memory hungry count distinct
- if (hasMemHungryCountDistinct == false) {
- return;
- }
-
- int rowSizeEst = dimensions.size() * 3;
- for (FunctionDesc func : metrics) {
- rowSizeEst += func.getReturnDataType().getSpaceEstimate();
- }
-
- long rowEst = MEM_BUDGET_PER_QUERY / rowSizeEst;
- context.setThreshold((int) rowEst);
- }
-
- private void setLimit(TupleFilter filter, StorageContext context) {
- boolean goodAggr = context.isExactAggregation();
- boolean goodFilter = filter == null || (TupleFilter.isEvaluableRecursively(filter) && context.isCoprocessorEnabled());
- boolean goodSort = context.hasSort() == false;
- if (goodAggr && goodFilter && goodSort) {
- logger.info("Enable limit " + context.getLimit());
- context.enableLimit();
- }
- }
-
- // ============================================================================
-
- @Override
- public Range<Long> getVolatilePeriod() {
- return null;
- }
-
- @Override
- public String getStorageUUID() {
- return cubeInstance.getUuid();
- }
-
- @Override
- public boolean isDynamic() {
- return false;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/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
deleted file mode 100644
index 993efbf..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeTupleConverter.java
+++ /dev/null
@@ -1,224 +0,0 @@
-/*
- * 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 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/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java b/storage/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java
deleted file mode 100644
index 096dba7..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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.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/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
deleted file mode 100644
index c42e89f..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/RawScan.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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/11611cc2/storage/src/main/java/org/apache/kylin/storage/cube/SequentialCubeTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/SequentialCubeTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/cube/SequentialCubeTupleIterator.java
deleted file mode 100644
index a55f1d0..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/SequentialCubeTupleIterator.java
+++ /dev/null
@@ -1,111 +0,0 @@
-package org.apache.kylin.storage.cube;
-
-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/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java
deleted file mode 100644
index e0605b4..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java
+++ /dev/null
@@ -1,37 +0,0 @@
-package org.apache.kylin.storage.cube;
-
-import java.util.List;
-
-import org.apache.kylin.metadata.tuple.ITuple;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-
-public class SerializedCubeTupleIterator implements ITupleIterator {
-
- public SerializedCubeTupleIterator(List<CubeScanner> scanners) {
- // TODO Auto-generated constructor stub
- }
-
- @Override
- public boolean hasNext() {
- // TODO Auto-generated method stub
- return false;
- }
-
- @Override
- public ITuple next() {
- // TODO Auto-generated method stub
- return null;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void close() {
- // TODO Auto-generated method stub
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/filter/BitMapFilterEvaluator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/filter/BitMapFilterEvaluator.java b/storage/src/main/java/org/apache/kylin/storage/filter/BitMapFilterEvaluator.java
deleted file mode 100644
index 9d9279f..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/filter/BitMapFilterEvaluator.java
+++ /dev/null
@@ -1,249 +0,0 @@
-/*
- * 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.filter;
-
-import java.util.List;
-
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.metadata.filter.CompareTupleFilter;
-import org.apache.kylin.metadata.filter.ConstantTupleFilter;
-import org.apache.kylin.metadata.filter.LogicalTupleFilter;
-import org.apache.kylin.metadata.filter.TupleFilter;
-import org.apache.kylin.metadata.model.TblColRef;
-
-import it.uniroma3.mat.extendedset.intset.ConciseSet;
-
-/**
- * @author yangli9
- *
- * Evaluate a group of records against a filter in batch.
- */
-public class BitMapFilterEvaluator {
-
- /** Provides bitmaps for a record group ranging [0..N-1], where N is the size of the group */
- public static interface BitMapProvider {
-
- /** return records whose specified column having specified value */
- ConciseSet getBitMap(TblColRef col, Integer startId, Integer endId);
-
- /** return the size of the group */
- int getRecordCount();
-
- /** return the max value ID of a column according to dictionary */
- int getMaxValueId(TblColRef col);
- }
-
- BitMapProvider provider;
-
- public BitMapFilterEvaluator(BitMapProvider bitMapProvider) {
- this.provider = bitMapProvider;
- }
-
- /**
- * @param filter
- * @return a set of records that match the filter; or null if filter is null or unable to evaluate
- */
- public ConciseSet evaluate(TupleFilter filter) {
- if (filter == null)
- return null;
-
- if (filter instanceof LogicalTupleFilter)
- return evalLogical((LogicalTupleFilter) filter);
-
- if (filter instanceof CompareTupleFilter)
- return evalCompare((CompareTupleFilter) filter);
-
- if (filter instanceof ConstantTupleFilter) {
- if (!filter.evaluate(null, null)) {
- return new ConciseSet();
- }
- }
-
- return null; // unable to evaluate
- }
-
- private ConciseSet evalCompare(CompareTupleFilter filter) {
- switch (filter.getOperator()) {
- case ISNULL:
- return evalCompareIsNull(filter);
- case ISNOTNULL:
- return evalCompareIsNotNull(filter);
- case EQ:
- return evalCompareEqual(filter);
- case NEQ:
- return evalCompareNotEqual(filter);
- case IN:
- return evalCompareIn(filter);
- case NOTIN:
- return evalCompareNotIn(filter);
- case LT:
- return evalCompareLT(filter);
- case LTE:
- return evalCompareLTE(filter);
- case GT:
- return evalCompareGT(filter);
- case GTE:
- return evalCompareGTE(filter);
- default:
- throw new IllegalStateException("Unsupported operator " + filter.getOperator());
- }
- }
-
- private ConciseSet evalCompareLT(CompareTupleFilter filter) {
- int id = Dictionary.stringToDictId((String) filter.getFirstValue());
- return collectRange(filter.getColumn(), null, id - 1);
- }
-
- private ConciseSet evalCompareLTE(CompareTupleFilter filter) {
- int id = Dictionary.stringToDictId((String) filter.getFirstValue());
- return collectRange(filter.getColumn(), null, id);
- }
-
- private ConciseSet evalCompareGT(CompareTupleFilter filter) {
- int id = Dictionary.stringToDictId((String) filter.getFirstValue());
- return collectRange(filter.getColumn(), id + 1, null);
- }
-
- private ConciseSet evalCompareGTE(CompareTupleFilter filter) {
- int id = Dictionary.stringToDictId((String) filter.getFirstValue());
- return collectRange(filter.getColumn(), id, null);
- }
-
- private ConciseSet collectRange(TblColRef column, Integer startId, Integer endId) {
- return provider.getBitMap(column, startId, endId);
- }
-
- private ConciseSet evalCompareEqual(CompareTupleFilter filter) {
- int id = Dictionary.stringToDictId((String) filter.getFirstValue());
- ConciseSet bitMap = provider.getBitMap(filter.getColumn(), id, id);
- if (bitMap == null)
- return null;
- return bitMap.clone(); // NOTE the clone() to void messing provider's cache
- }
-
- private ConciseSet evalCompareNotEqual(CompareTupleFilter filter) {
- ConciseSet set = evalCompareEqual(filter);
- not(set);
- dropNull(set, filter);
- return set;
- }
-
- private ConciseSet evalCompareIn(CompareTupleFilter filter) {
- ConciseSet set = new ConciseSet();
- for (Object value : filter.getValues()) {
- int id = Dictionary.stringToDictId((String) value);
- ConciseSet bitMap = provider.getBitMap(filter.getColumn(), id, id);
- if (bitMap == null)
- return null;
- set.addAll(bitMap);
- }
- return set;
- }
-
- private ConciseSet evalCompareNotIn(CompareTupleFilter filter) {
- ConciseSet set = evalCompareIn(filter);
- not(set);
- dropNull(set, filter);
- return set;
- }
-
- private void dropNull(ConciseSet set, CompareTupleFilter filter) {
- if (set == null)
- return;
-
- ConciseSet nullSet = evalCompareIsNull(filter);
- set.removeAll(nullSet);
- }
-
- private ConciseSet evalCompareIsNull(CompareTupleFilter filter) {
- ConciseSet bitMap = provider.getBitMap(filter.getColumn(), null, null);
- if (bitMap == null)
- return null;
- return bitMap.clone(); // NOTE the clone() to void messing provider's cache
- }
-
- private ConciseSet evalCompareIsNotNull(CompareTupleFilter filter) {
- ConciseSet set = evalCompareIsNull(filter);
- not(set);
- return set;
- }
-
- private ConciseSet evalLogical(LogicalTupleFilter filter) {
- List<? extends TupleFilter> children = filter.getChildren();
-
- switch (filter.getOperator()) {
- case AND:
- return evalLogicalAnd(children);
- case OR:
- return evalLogicalOr(children);
- case NOT:
- return evalLogicalNot(children);
- default:
- throw new IllegalStateException("Unsupported operator " + filter.getOperator());
- }
- }
-
- private ConciseSet evalLogicalAnd(List<? extends TupleFilter> children) {
- ConciseSet set = new ConciseSet();
- not(set);
-
- for (TupleFilter c : children) {
- ConciseSet t = evaluate(c);
- if (t == null)
- continue; // because it's AND
-
- set.retainAll(t);
- }
- return set;
- }
-
- private ConciseSet evalLogicalOr(List<? extends TupleFilter> children) {
- ConciseSet set = new ConciseSet();
-
- for (TupleFilter c : children) {
- ConciseSet t = evaluate(c);
- if (t == null)
- return null; // because it's OR
-
- set.addAll(t);
- }
- return set;
- }
-
- private ConciseSet evalLogicalNot(List<? extends TupleFilter> children) {
- ConciseSet set = evaluate(children.get(0));
- not(set);
- return set;
- }
-
- private void not(ConciseSet set) {
- if (set == null)
- return;
-
- set.add(provider.getRecordCount());
- set.complement();
- }
-
- public static void main(String[] args) {
- ConciseSet s = new ConciseSet();
- s.add(5);
- s.complement();
- System.out.println(s);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/ColumnValueRange.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/ColumnValueRange.java b/storage/src/main/java/org/apache/kylin/storage/hbase/ColumnValueRange.java
deleted file mode 100644
index 1a594e1..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/ColumnValueRange.java
+++ /dev/null
@@ -1,214 +0,0 @@
-/*
- * 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;
-
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Set;
-
-import org.apache.kylin.cube.kv.RowKeyColumnOrder;
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
-import org.apache.kylin.metadata.model.TblColRef;
-
-import com.google.common.collect.Sets;
-
-/**
- *
- * @author xjiang
- *
- */
-public class ColumnValueRange {
- private TblColRef column;
- private RowKeyColumnOrder order;
- private String beginValue;
- private String endValue;
- private Set<String> equalValues;
-
- public ColumnValueRange(TblColRef column, Collection<String> values, FilterOperatorEnum op) {
- this.column = column;
- this.order = RowKeyColumnOrder.getInstance(column.getType());
-
- switch (op) {
- case EQ:
- case IN:
- equalValues = new HashSet<String>(values);
- refreshBeginEndFromEquals();
- break;
- case LT:
- case LTE:
- endValue = order.max(values);
- break;
- case GT:
- case GTE:
- beginValue = order.min(values);
- break;
- case NEQ:
- case NOTIN:
- case ISNULL: // TODO ISNULL worth pass down as a special equal value
- case ISNOTNULL:
- // let Optiq filter it!
- break;
- default:
- throw new UnsupportedOperationException(op.name());
- }
- }
-
- public ColumnValueRange(TblColRef column, String beginValue, String endValue, Set<String> equalValues) {
- copy(column, beginValue, endValue, equalValues);
- }
-
- void copy(TblColRef column, String beginValue, String endValue, Set<String> equalValues) {
- this.column = column;
- this.order = RowKeyColumnOrder.getInstance(column.getType());
- this.beginValue = beginValue;
- this.endValue = endValue;
- this.equalValues = equalValues;
- }
-
- public TblColRef getColumn() {
- return column;
- }
-
- public String getBeginValue() {
- return beginValue;
- }
-
- public String getEndValue() {
- return endValue;
- }
-
- public Set<String> getEqualValues() {
- return equalValues;
- }
-
- private void refreshBeginEndFromEquals() {
- this.beginValue = order.min(this.equalValues);
- this.endValue = order.max(this.equalValues);
- }
-
- public boolean satisfyAll() {
- return beginValue == null && endValue == null; // the NEQ case
- }
-
- public boolean satisfyNone() {
- if (equalValues != null) {
- return equalValues.isEmpty();
- } else if (beginValue != null && endValue != null) {
- return order.compare(beginValue, endValue) > 0;
- } else {
- return false;
- }
- }
-
- public void andMerge(ColumnValueRange another) {
- assert this.column.equals(another.column);
-
- if (another.satisfyAll()) {
- return;
- }
-
- if (this.satisfyAll()) {
- copy(another.column, another.beginValue, another.endValue, another.equalValues);
- return;
- }
-
- if (this.equalValues != null && another.equalValues != null) {
- this.equalValues.retainAll(another.equalValues);
- refreshBeginEndFromEquals();
- return;
- }
-
- if (this.equalValues != null) {
- this.equalValues = filter(this.equalValues, another.beginValue, another.endValue);
- refreshBeginEndFromEquals();
- return;
- }
-
- if (another.equalValues != null) {
- this.equalValues = filter(another.equalValues, this.beginValue, this.endValue);
- refreshBeginEndFromEquals();
- return;
- }
-
- this.beginValue = order.max(this.beginValue, another.beginValue);
- this.endValue = order.min(this.endValue, another.endValue);
- }
-
- private Set<String> filter(Set<String> equalValues, String beginValue, String endValue) {
- Set<String> result = Sets.newHashSetWithExpectedSize(equalValues.size());
- for (String v : equalValues) {
- if (between(v, beginValue, endValue)) {
- result.add(v);
- }
- }
- return equalValues;
- }
-
- private boolean between(String v, String beginValue, String endValue) {
- return (beginValue == null || order.compare(beginValue, v) <= 0) && (endValue == null || order.compare(v, endValue) <= 0);
- }
-
- // remove invalid EQ/IN values and round start/end according to dictionary
- public void preEvaluateWithDict(Dictionary<String> dict) {
- if (dict == null)
- return;
-
- if (equalValues != null) {
- Iterator<String> it = equalValues.iterator();
- while (it.hasNext()) {
- String v = it.next();
- try {
- dict.getIdFromValue(v);
- } catch (IllegalArgumentException e) {
- // value not in dictionary
- it.remove();
- }
- }
- refreshBeginEndFromEquals();
- }
-
- if (beginValue != null) {
- try {
- beginValue = dict.getValueFromId(dict.getIdFromValue(beginValue, -1));
- } catch (IllegalArgumentException e) {
- // value is less than the smallest in dictionary
- beginValue = null;
- }
- }
-
- if (endValue != null) {
- try {
- endValue = dict.getValueFromId(dict.getIdFromValue(endValue, 1));
- } catch (IllegalArgumentException e) {
- // value is greater than the biggest in dictionary
- endValue = null;
- }
- }
- }
-
- public String toString() {
- if (equalValues == null) {
- return column.getName() + " between " + beginValue + " and " + endValue;
- } else {
- return column.getName() + " in " + equalValues;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java
deleted file mode 100644
index 987ebb9..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java
+++ /dev/null
@@ -1,277 +0,0 @@
-/*
- * 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;
-
-import java.text.MessageFormat;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.Set;
-
-import org.apache.hadoop.hbase.client.HConnection;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
-import org.apache.hadoop.hbase.filter.Filter;
-import org.apache.hadoop.hbase.filter.FilterList;
-import org.apache.hadoop.hbase.filter.FuzzyRowFilter;
-import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
-import org.apache.kylin.common.persistence.StorageException;
-import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.cube.model.HBaseColumnDesc;
-import org.apache.kylin.metadata.filter.TupleFilter;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.storage.StorageContext;
-import org.apache.kylin.storage.hbase.coprocessor.observer.ObserverEnabler;
-import org.apache.kylin.storage.hbase.steps.RowValueDecoder;
-import org.apache.kylin.storage.tuple.Tuple;
-import org.apache.kylin.storage.tuple.TupleInfo;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Lists;
-
-/**
- * @author xjiang
- *
- */
-public class CubeSegmentTupleIterator implements ITupleIterator {
-
- public static final Logger logger = LoggerFactory.getLogger(CubeSegmentTupleIterator.class);
-
- public static final int SCAN_CACHE = 1024;
-
- private final CubeSegment cubeSeg;
- private final TupleFilter filter;
- private final Collection<TblColRef> groupBy;
- private final Collection<RowValueDecoder> rowValueDecoders;
- private final StorageContext context;
- private final String tableName;
- private final HTableInterface table;
-
- private final CubeTupleConverter tupleConverter;
- private final Iterator<HBaseKeyRange> rangeIterator;
- private final Tuple oneTuple; // avoid new instance
-
- private Scan scan;
- private ResultScanner scanner;
- private Iterator<Result> resultIterator;
- private int scanCount;
- private Tuple next;
-
- public CubeSegmentTupleIterator(CubeSegment cubeSeg, List<HBaseKeyRange> keyRanges, HConnection conn, //
- Set<TblColRef> dimensions, TupleFilter filter, Set<TblColRef> groupBy, //
- List<RowValueDecoder> rowValueDecoders, StorageContext context, TupleInfo returnTupleInfo) {
- this.cubeSeg = cubeSeg;
- this.filter = filter;
- this.groupBy = groupBy;
- this.rowValueDecoders = rowValueDecoders;
- this.context = context;
- this.tableName = cubeSeg.getStorageLocationIdentifier();
-
- Cuboid cuboid = keyRanges.get(0).getCuboid();
- for (HBaseKeyRange range : keyRanges) {
- assert cuboid.equals(range.getCuboid());
- }
-
- this.tupleConverter = new CubeTupleConverter(cubeSeg, cuboid, rowValueDecoders, returnTupleInfo);
- this.oneTuple = new Tuple(returnTupleInfo);
- this.rangeIterator = keyRanges.iterator();
- this.scanCount = 0;
-
- try {
- this.table = conn.getTable(tableName);
- } catch (Throwable t) {
- throw new StorageException("Error when open connection to table " + tableName, t);
- }
- }
-
- @Override
- public boolean hasNext() {
- if (next != null)
- return true;
-
- if (resultIterator == null) {
- if (rangeIterator.hasNext() == false)
- return false;
-
- resultIterator = doScan(rangeIterator.next());
- }
-
- if (resultIterator.hasNext() == false) {
- closeScanner();
- resultIterator = null;
- return hasNext();
- }
-
- Result result = resultIterator.next();
- scanCount++;
- tupleConverter.translateResult(result, oneTuple);
- next = oneTuple;
- return true;
- }
-
- @Override
- public Tuple next() {
- if (next == null) {
- hasNext();
- if (next == null)
- throw new NoSuchElementException();
- }
- Tuple r = next;
- next = null;
- return r;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-
- private final Iterator<Result> doScan(HBaseKeyRange keyRange) {
- Iterator<Result> iter = null;
- try {
- scan = buildScan(keyRange);
- applyFuzzyFilter(scan, keyRange);
- logScan(keyRange);
-
- scanner = ObserverEnabler.scanWithCoprocessorIfBeneficial(cubeSeg, keyRange.getCuboid(), filter, groupBy, rowValueDecoders, context, table, scan);
-
- iter = scanner.iterator();
- } catch (Throwable t) {
- String msg = MessageFormat.format("Error when scan from lower key {1} to upper key {2} on table {0}.", tableName, Bytes.toString(keyRange.getStartKey()), Bytes.toString(keyRange.getStopKey()));
- throw new StorageException(msg, t);
- }
- return iter;
- }
-
- private void logScan(HBaseKeyRange keyRange) {
- StringBuilder info = new StringBuilder();
- info.append("\nScan hbase table ").append(tableName).append(": ");
- if (keyRange.getCuboid().requirePostAggregation()) {
- info.append("cuboid require post aggregation, from ");
- } else {
- info.append("cuboid exact match, from ");
- }
- info.append(keyRange.getCuboid().getInputID());
- info.append(" to ");
- info.append(keyRange.getCuboid().getId());
- info.append("\nStart: ");
- info.append(keyRange.getStartKeyAsString());
- info.append(" - ");
- info.append(Bytes.toStringBinary(keyRange.getStartKey()));
- info.append("\nStop: ");
- info.append(keyRange.getStopKeyAsString());
- info.append(" - ");
- info.append(Bytes.toStringBinary(keyRange.getStopKey()));
- if (this.scan.getFilter() != null) {
- info.append("\nFuzzy: ");
- info.append(keyRange.getFuzzyKeyAsString());
- }
- logger.info(info.toString());
- }
-
- private Scan buildScan(HBaseKeyRange keyRange) {
- Scan scan = new Scan();
- scan.setCaching(SCAN_CACHE);
- scan.setCacheBlocks(true);
- scan.setAttribute(Scan.SCAN_ATTRIBUTES_METRICS_ENABLE, Bytes.toBytes(Boolean.TRUE));
- for (RowValueDecoder valueDecoder : this.rowValueDecoders) {
- HBaseColumnDesc hbaseColumn = valueDecoder.getHBaseColumn();
- byte[] byteFamily = Bytes.toBytes(hbaseColumn.getColumnFamilyName());
- byte[] byteQualifier = Bytes.toBytes(hbaseColumn.getQualifier());
- scan.addColumn(byteFamily, byteQualifier);
- }
- scan.setStartRow(keyRange.getStartKey());
- scan.setStopRow(keyRange.getStopKey());
- return scan;
- }
-
- private void applyFuzzyFilter(Scan scan, HBaseKeyRange keyRange) {
- List<org.apache.kylin.common.util.Pair<byte[], byte[]>> fuzzyKeys = keyRange.getFuzzyKeys();
- 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 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;
- }
-
- private void closeScanner() {
- if (logger.isDebugEnabled() && scan != null) {
- logger.debug("Scan " + scan.toString());
- byte[] metricsBytes = scan.getAttribute(Scan.SCAN_ATTRIBUTES_METRICS_DATA);
- if (metricsBytes != null) {
- ScanMetrics scanMetrics = ProtobufUtil.toScanMetrics(metricsBytes);
- logger.debug("HBase Metrics: " + "count={}, ms={}, bytes={}, remote_bytes={}, regions={}, not_serving_region={}, rpc={}, rpc_retries={}, remote_rpc={}, remote_rpc_retries={}", //
- new Object[] { scanCount, scanMetrics.sumOfMillisSecBetweenNexts, scanMetrics.countOfBytesInResults, scanMetrics.countOfBytesInRemoteResults, scanMetrics.countOfRegions, scanMetrics.countOfNSRE, scanMetrics.countOfRPCcalls, scanMetrics.countOfRPCRetries, scanMetrics.countOfRemoteRPCcalls, scanMetrics.countOfRemoteRPCRetries });
- }
- }
- try {
- if (scanner != null) {
- scanner.close();
- scanner = null;
- }
- } catch (Throwable t) {
- throw new StorageException("Error when close scanner for table " + tableName, t);
- }
- }
-
- private void closeTable() {
- try {
- if (table != null) {
- table.close();
- }
- } catch (Throwable t) {
- throw new StorageException("Error when close table " + tableName, t);
- }
- }
-
- @Override
- public void close() {
- logger.info("Closing CubeSegmentTupleIterator");
- closeScanner();
- closeTable();
- }
-
-}