You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2015/08/24 11:00:45 UTC
[06/22] incubator-kylin git commit: KYLIN-960 organize existing
storage module
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/3d2685d8/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/EndpointTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/EndpointTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/EndpointTupleIterator.java
deleted file mode 100644
index c33f2e4..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/EndpointTupleIterator.java
+++ /dev/null
@@ -1,363 +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.coprocessor.endpoint;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-
-import javax.annotation.Nullable;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.SerializationUtils;
-import org.apache.hadoop.hbase.client.HConnection;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.coprocessor.Batch;
-import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
-import org.apache.hadoop.hbase.ipc.ServerRpcController;
-import org.apache.kylin.common.util.CompressionUtils;
-import org.apache.kylin.common.util.DateFormat;
-import org.apache.kylin.common.util.RangeUtil;
-import org.apache.kylin.invertedindex.IISegment;
-import org.apache.kylin.invertedindex.index.TableRecord;
-import org.apache.kylin.invertedindex.index.TableRecordInfo;
-import org.apache.kylin.metadata.filter.ConstantTupleFilter;
-import org.apache.kylin.metadata.filter.TupleFilter;
-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.StorageContext;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorFilter;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorProjector;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorRowType;
-import org.apache.kylin.storage.hbase.coprocessor.FilterDecorator;
-import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos;
-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.base.Function;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Collections2;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
-import com.google.common.collect.Sets;
-import com.google.protobuf.ByteString;
-
-/**
- */
-public class EndpointTupleIterator implements ITupleIterator {
-
- private final static Logger logger = LoggerFactory.getLogger(EndpointTupleIterator.class);
-
- private final IISegment seg;
-
- private final String factTableName;
- private final List<TblColRef> columns;
- private final TupleInfo tupleInfo;
- private final TableRecordInfo tableRecordInfo;
- private final EndpointTupleConverter tupleConverter;
-
- private final CoprocessorRowType pushedDownRowType;
- private final CoprocessorFilter pushedDownFilter;
- private final CoprocessorProjector pushedDownProjector;
- private final EndpointAggregators pushedDownAggregators;
- private final Range<Long> tsRange;//timestamp column condition's interval
-
- private Iterator<List<IIProtos.IIResponseInternal.IIRow>> regionResponsesIterator = null;
- private ITupleIterator tupleIterator = null;
- private HTableInterface table = null;
-
- private TblColRef partitionCol;
- private long lastDataTime = -1;
- private int rowsInAllMetric = 0;
-
- public EndpointTupleIterator(IISegment segment, TupleFilter rootFilter, Collection<TblColRef> groupBy, List<FunctionDesc> measures, StorageContext context, HConnection conn, TupleInfo returnTupleInfo) throws Throwable {
-
- String tableName = segment.getStorageLocationIdentifier();
- table = conn.getTable(tableName);
- factTableName = segment.getIIDesc().getFactTableName();
-
- if (rootFilter == null) {
- rootFilter = ConstantTupleFilter.TRUE;
- }
-
- if (groupBy == null) {
- groupBy = Sets.newHashSet();
- }
-
- if (measures == null) {
- measures = Lists.newArrayList();
- }
-
- //this method will change measures
- rewriteMeasureParameters(measures, segment.getColumns());
-
- this.seg = segment;
- this.columns = segment.getColumns();
-
- this.tupleInfo = returnTupleInfo;
- this.tupleConverter = new EndpointTupleConverter(columns, measures, returnTupleInfo);
- this.tableRecordInfo = new TableRecordInfo(this.seg);
-
- this.pushedDownRowType = CoprocessorRowType.fromTableRecordInfo(tableRecordInfo, this.columns);
- this.pushedDownFilter = CoprocessorFilter.fromFilter(new ClearTextDictionary(this.tableRecordInfo), rootFilter, FilterDecorator.FilterConstantsTreatment.AS_IT_IS);
-
- for (TblColRef column : this.pushedDownFilter.getInevaluableColumns()) {
- groupBy.add(column);
- }
-
- this.pushedDownProjector = CoprocessorProjector.makeForEndpoint(tableRecordInfo, groupBy);
- this.pushedDownAggregators = EndpointAggregators.fromFunctions(tableRecordInfo, measures);
-
- int tsCol = this.tableRecordInfo.getTimestampColumn();
- this.partitionCol = this.columns.get(tsCol);
- this.tsRange = TsConditionExtractor.extractTsCondition(this.partitionCol, rootFilter);
-
- if (this.tsRange == null) {
- logger.info("TsRange conflict for endpoint, return empty directly");
- this.tupleIterator = ITupleIterator.EMPTY_TUPLE_ITERATOR;
- } else {
- logger.info("The tsRange being pushed is " + RangeUtil.formatTsRange(tsRange));
- }
-
- IIProtos.IIRequest endpointRequest = prepareRequest();
- Collection<IIProtos.IIResponse> compressedShardResults = getResults(endpointRequest, table);
-
- //decompress
- Collection<IIProtos.IIResponseInternal> shardResults = new ArrayList<>();
- for (IIProtos.IIResponse input : compressedShardResults) {
- byte[] compressed = input.getBlob().toByteArray();
- try {
- byte[] decompressed = CompressionUtils.decompress(compressed);
- shardResults.add(IIProtos.IIResponseInternal.parseFrom(decompressed));
- } catch (Exception e) {
- throw new RuntimeException("decompress endpoint response error");
- }
- }
-
- this.lastDataTime = Collections.min(Collections2.transform(shardResults, new Function<IIProtos.IIResponseInternal, Long>() {
- @Nullable
- @Override
- public Long apply(IIProtos.IIResponseInternal input) {
-
- IIProtos.IIResponseInternal.Stats status = input.getStats();
- logger.info("Endpoints all returned, stats from shard {}: start moment:{}, finish moment: {}, elapsed ms: {}, scanned slices: {}, latest slice time is {}",//
- new Object[] { String.valueOf(status.getMyShard()),//
- DateFormat.formatToTimeStr(status.getServiceStartTime()),//
- DateFormat.formatToTimeStr(status.getServiceEndTime()),//
- String.valueOf(status.getServiceEndTime() - status.getServiceStartTime()),//
- String.valueOf(status.getScannedSlices()), DateFormat.formatToTimeStr(status.getLatestDataTime()) });
-
- return status.getLatestDataTime();
- }
- }));
-
- this.regionResponsesIterator = Collections2.transform(shardResults, new Function<IIProtos.IIResponseInternal, List<IIProtos.IIResponseInternal.IIRow>>() {
- @Nullable
- @Override
- public List<IIProtos.IIResponseInternal.IIRow> apply(@Nullable IIProtos.IIResponseInternal input) {
- return input.getRowsList();
- }
- }).iterator();
-
- if (this.regionResponsesIterator.hasNext()) {
- this.tupleIterator = new SingleRegionTupleIterator(this.regionResponsesIterator.next());
- } else {
- this.tupleIterator = ITupleIterator.EMPTY_TUPLE_ITERATOR;
- }
- }
-
- /**
- * measure comes from query engine, does not contain enough information
- */
- private void rewriteMeasureParameters(List<FunctionDesc> measures, List<TblColRef> columns) {
- for (FunctionDesc functionDesc : measures) {
- if (functionDesc.isCount()) {
- functionDesc.setReturnType("bigint");
- } else {
- boolean updated = false;
- for (TblColRef column : columns) {
- if (column.isSameAs(factTableName, functionDesc.getParameter().getValue())) {
- if (functionDesc.isCountDistinct()) {
- //TODO: default precision might need be configurable
- String iiDefaultHLLC = "hllc10";
- functionDesc.setReturnType(iiDefaultHLLC);
- } else {
- functionDesc.setReturnType(column.getColumnDesc().getType().toString());
- }
- functionDesc.getParameter().setColRefs(ImmutableList.of(column));
- updated = true;
- break;
- }
- }
- if (!updated) {
- throw new RuntimeException("Func " + functionDesc + " is not related to any column in fact table " + factTableName);
- }
- }
- }
- }
-
- @Override
- public boolean hasNext() {
- while (!this.tupleIterator.hasNext()) {
- if (this.regionResponsesIterator.hasNext()) {
- this.tupleIterator = new SingleRegionTupleIterator(this.regionResponsesIterator.next());
- } else {
- return false;
- }
- }
- return true;
- }
-
- @Override
- public ITuple next() {
- rowsInAllMetric++;
-
- if (!hasNext()) {
- throw new IllegalStateException("No more ITuple in EndpointTupleIterator");
- }
-
- ITuple tuple = this.tupleIterator.next();
- return tuple;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
-
- }
-
- @Override
- public void close() {
- IOUtils.closeQuietly(table);
- logger.info("Closed after " + rowsInAllMetric + " rows are fetched");
- }
-
- /**
- * tells storage layer cache what time period of data should not be cached.
- * for static storage like cube, it will return null
- * for dynamic storage like ii, it will for example exclude the last two minutes for possible data latency
- * @return
- */
- public Range<Long> getCacheExcludedPeriod() {
- Preconditions.checkArgument(lastDataTime != -1, "lastDataTime is not set yet");
- return Ranges.greaterThan(lastDataTime);
- }
-
- private IIProtos.IIRequest prepareRequest() throws IOException {
- IIProtos.IIRequest.Builder builder = IIProtos.IIRequest.newBuilder();
-
- if (this.tsRange != null) {
- byte[] tsRangeBytes = SerializationUtils.serialize(this.tsRange);
- builder.setTsRange(ByteString.copyFrom(tsRangeBytes));
- }
-
- builder.setType(ByteString.copyFrom(CoprocessorRowType.serialize(pushedDownRowType))) //
- .setFilter(ByteString.copyFrom(CoprocessorFilter.serialize(pushedDownFilter))) //
- .setProjector(ByteString.copyFrom(CoprocessorProjector.serialize(pushedDownProjector))) //
- .setAggregator(ByteString.copyFrom(EndpointAggregators.serialize(pushedDownAggregators)));
-
- IIProtos.IIRequest request = builder.build();
-
- return request;
- }
-
- //TODO : async callback
- private Collection<IIProtos.IIResponse> getResults(final IIProtos.IIRequest request, HTableInterface table) throws Throwable {
- Map<byte[], IIProtos.IIResponse> results = table.coprocessorService(IIProtos.RowsService.class, null, null, new Batch.Call<IIProtos.RowsService, IIProtos.IIResponse>() {
- public IIProtos.IIResponse call(IIProtos.RowsService rowsService) throws IOException {
- ServerRpcController controller = new ServerRpcController();
- BlockingRpcCallback<IIProtos.IIResponse> rpcCallback = new BlockingRpcCallback<>();
- rowsService.getRows(controller, request, rpcCallback);
- IIProtos.IIResponse response = rpcCallback.get();
- if (controller.failedOnException()) {
- throw controller.getFailedOn();
- }
-
- return response;
- }
- });
-
- return results.values();
- }
-
- /**
- * Internal class to handle iterators for a single region's returned rows
- */
- class SingleRegionTupleIterator implements ITupleIterator {
- private List<IIProtos.IIResponseInternal.IIRow> rows;
- private int index = 0;
-
- //not thread safe!
- private TableRecord tableRecord;
- private List<Object> measureValues;
- private Tuple tuple;
-
- public SingleRegionTupleIterator(List<IIProtos.IIResponseInternal.IIRow> rows) {
- this.rows = rows;
- this.index = 0;
- this.tableRecord = tableRecordInfo.createTableRecord();
- this.tuple = new Tuple(tupleInfo);
- }
-
- @Override
- public boolean hasNext() {
- return index < rows.size();
- }
-
- @Override
- public ITuple next() {
- if (!hasNext()) {
- throw new IllegalStateException("No more Tuple in the SingleRegionTupleIterator");
- }
-
- IIProtos.IIResponseInternal.IIRow currentRow = rows.get(index);
- byte[] columnsBytes = currentRow.getColumns().toByteArray();
- this.tableRecord.setBytes(columnsBytes, 0, columnsBytes.length);
- if (currentRow.hasMeasures()) {
- byte[] measuresBytes = currentRow.getMeasures().toByteArray();
-
- this.measureValues = pushedDownAggregators.deserializeMetricValues(measuresBytes, 0);
- }
-
- index++;
-
- return tupleConverter.makeTuple(this.tableRecord, this.measureValues, this.tuple);
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void close() {
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/3d2685d8/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/HbaseServerKVIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/HbaseServerKVIterator.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/HbaseServerKVIterator.java
deleted file mode 100644
index c186f5e..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/HbaseServerKVIterator.java
+++ /dev/null
@@ -1,104 +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.coprocessor.endpoint;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.kylin.invertedindex.model.IIRow;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Lists;
-
-/**
- */
-public class HbaseServerKVIterator implements Iterable<IIRow>, Closeable {
-
- private RegionScanner innerScanner;
- private Logger logger = LoggerFactory.getLogger(HbaseServerKVIterator.class);
-
- public HbaseServerKVIterator(RegionScanner innerScanner) {
- this.innerScanner = innerScanner;
- }
-
- @Override
- public void close() throws IOException {
- IOUtils.closeQuietly(this.innerScanner);
- }
-
- private static class IIRowIterator implements Iterator<IIRow> {
-
- private final RegionScanner regionScanner;
- private final IIRow row = new IIRow();
- List<Cell> results = Lists.newArrayList();
-
- private boolean hasMore;
-
- IIRowIterator(RegionScanner innerScanner) {
- this.regionScanner = innerScanner;
- try {
- hasMore = regionScanner.nextRaw(results);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public boolean hasNext() {
- return !results.isEmpty();
- }
-
- @Override
- public IIRow next() {
- if (results.size() < 1) {
- throw new NoSuchElementException();
- }
- for (Cell c : results) {
- row.updateWith(c);
- }
- results.clear();
- try {
- if (hasMore) {
- hasMore = regionScanner.nextRaw(results);
- }
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- return row;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
- }
-
- @Override
- public Iterator<IIRow> iterator() {
- return new IIRowIterator(innerScanner);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/3d2685d8/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
deleted file mode 100644
index 9f9e79f..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
+++ /dev/null
@@ -1,319 +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.coprocessor.endpoint;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.Map;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang.ArrayUtils;
-import org.apache.commons.lang.SerializationUtils;
-import org.apache.hadoop.hbase.Coprocessor;
-import org.apache.hadoop.hbase.CoprocessorEnvironment;
-import org.apache.hadoop.hbase.client.Result;
-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.Array;
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.common.util.CompressionUtils;
-import org.apache.kylin.cube.kv.RowKeyColumnIO;
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.dict.TrieDictionary;
-import org.apache.kylin.invertedindex.index.RawTableRecord;
-import org.apache.kylin.invertedindex.index.Slice;
-import org.apache.kylin.invertedindex.index.TableRecordInfoDigest;
-import org.apache.kylin.invertedindex.model.IIDesc;
-import org.apache.kylin.invertedindex.model.IIKeyValueCodec;
-import org.apache.kylin.metadata.measure.MeasureAggregator;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.storage.filter.BitMapFilterEvaluator;
-import org.apache.kylin.storage.hbase.coprocessor.AggrKey;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorConstants;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorFilter;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorProjector;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorRowType;
-import org.apache.kylin.storage.hbase.coprocessor.FilterDecorator;
-import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Range;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.RpcCallback;
-import com.google.protobuf.RpcController;
-import com.google.protobuf.Service;
-import it.uniroma3.mat.extendedset.intset.ConciseSet;
-
-/**
- */
-public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, CoprocessorService {
-
- private static final Logger logger = LoggerFactory.getLogger(IIEndpoint.class);
- private static final int MEMORY_LIMIT = 500 * 1024 * 1024;
-
- private RegionCoprocessorEnvironment env;
- private long serviceStartTime;
- private int shard;
-
- public IIEndpoint() {
- }
-
- private Scan prepareScan(IIProtos.IIRequest request, HRegion region) throws IOException {
- Scan scan = new Scan();
-
- scan.addColumn(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_QUALIFIER_BYTES);
- scan.addColumn(IIDesc.HBASE_FAMILY_BYTES, IIDesc.HBASE_DICTIONARY_BYTES);
-
- if (request.hasTsRange()) {
- Range<Long> tsRange = (Range<Long>) SerializationUtils.deserialize(request.getTsRange().toByteArray());
- byte[] regionStartKey = region.getStartKey();
- if (!ArrayUtils.isEmpty(regionStartKey)) {
- shard = BytesUtil.readUnsigned(regionStartKey, 0, IIKeyValueCodec.SHARD_LEN);
- } else {
- shard = 0;
- }
- logger.info("Start key of the region is: " + BytesUtil.toReadableText(regionStartKey) + ", making shard to be :" + shard);
-
- if (tsRange.hasLowerBound()) {
- //differentiate GT and GTE seems not very beneficial
- Preconditions.checkArgument(shard != -1, "Shard is -1!");
- long tsStart = tsRange.lowerEndpoint();
- logger.info("ts start is " + tsStart);
-
- byte[] idealStartKey = new byte[IIKeyValueCodec.SHARD_LEN + IIKeyValueCodec.TIMEPART_LEN];
- BytesUtil.writeUnsigned(shard, idealStartKey, 0, IIKeyValueCodec.SHARD_LEN);
- BytesUtil.writeLong(tsStart, idealStartKey, IIKeyValueCodec.SHARD_LEN, IIKeyValueCodec.TIMEPART_LEN);
- logger.info("ideaStartKey is(readable) :" + BytesUtil.toReadableText(idealStartKey));
- Result result = region.getClosestRowBefore(idealStartKey, IIDesc.HBASE_FAMILY_BYTES);
- if (result != null) {
- byte[] actualStartKey = Arrays.copyOf(result.getRow(), IIKeyValueCodec.SHARD_LEN + IIKeyValueCodec.TIMEPART_LEN);
- scan.setStartRow(actualStartKey);
- logger.info("The start key is set to " + BytesUtil.toReadableText(actualStartKey));
- } else {
- logger.info("There is no key before ideaStartKey so ignore tsStart");
- }
- }
-
- if (tsRange.hasUpperBound()) {
- //differentiate LT and LTE seems not very beneficial
- Preconditions.checkArgument(shard != -1, "Shard is -1");
- long tsEnd = tsRange.upperEndpoint();
- logger.info("ts end is " + tsEnd);
-
- byte[] actualEndKey = new byte[IIKeyValueCodec.SHARD_LEN + IIKeyValueCodec.TIMEPART_LEN];
- BytesUtil.writeUnsigned(shard, actualEndKey, 0, IIKeyValueCodec.SHARD_LEN);
- BytesUtil.writeLong(tsEnd + 1, actualEndKey, IIKeyValueCodec.SHARD_LEN, IIKeyValueCodec.TIMEPART_LEN);//notice +1 here
- scan.setStopRow(actualEndKey);
- logger.info("The stop key is set to " + BytesUtil.toReadableText(actualEndKey));
- }
- }
-
- return scan;
- }
-
- @Override
- public void getRows(RpcController controller, IIProtos.IIRequest request, RpcCallback<IIProtos.IIResponse> done) {
-
- this.serviceStartTime = System.currentTimeMillis();
-
- RegionScanner innerScanner = null;
- HRegion region = null;
-
- try {
- region = env.getRegion();
- region.startRegionOperation();
-
- innerScanner = region.getScanner(prepareScan(request, region));
-
- CoprocessorRowType type = CoprocessorRowType.deserialize(request.getType().toByteArray());
- CoprocessorProjector projector = CoprocessorProjector.deserialize(request.getProjector().toByteArray());
- EndpointAggregators aggregators = EndpointAggregators.deserialize(request.getAggregator().toByteArray());
- CoprocessorFilter filter = CoprocessorFilter.deserialize(request.getFilter().toByteArray());
-
- //compression
- IIProtos.IIResponseInternal response = getResponse(innerScanner, type, projector, aggregators, filter);
- byte[] compressed = CompressionUtils.compress(response.toByteArray());
- IIProtos.IIResponse compressedR = IIProtos.IIResponse.newBuilder().setBlob(ByteString.copyFrom(compressed)).build();
-
- done.run(compressedR);
- } 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);
- }
- }
- }
- }
-
- public IIProtos.IIResponseInternal getResponse(RegionScanner innerScanner, CoprocessorRowType type, CoprocessorProjector projector, EndpointAggregators aggregators, CoprocessorFilter filter) {
-
- TableRecordInfoDigest tableRecordInfoDigest = aggregators.getTableRecordInfoDigest();
-
- IIProtos.IIResponseInternal response;
-
- synchronized (innerScanner) {
- IIKeyValueCodec codec = new IIKeyValueCodec(tableRecordInfoDigest);
- //TODO pass projector to codec to skip loading columns
- Iterable<Slice> slices = codec.decodeKeyValue(new HbaseServerKVIterator(innerScanner));
-
- response = getResponseInternal(slices, tableRecordInfoDigest, filter, type, projector, aggregators);
- }
- return response;
- }
-
- private IIProtos.IIResponseInternal getResponseInternal(Iterable<Slice> slices, TableRecordInfoDigest recordInfo, CoprocessorFilter filter, CoprocessorRowType type, CoprocessorProjector projector, EndpointAggregators aggregators) {
- boolean needAgg = projector.hasGroupby() || !aggregators.isEmpty();
-
- //for needAgg use
- EndpointAggregationCache aggCache = new EndpointAggregationCache(aggregators);
- //for no needAgg use
- final int byteFormLen = recordInfo.getByteFormLen();
- int totalByteFormLen = 0;
-
- IIProtos.IIResponseInternal.Builder responseBuilder = IIProtos.IIResponseInternal.newBuilder();
- ClearTextDictionary clearTextDictionary = new ClearTextDictionary(recordInfo, type);
- RowKeyColumnIO rowKeyColumnIO = new RowKeyColumnIO(clearTextDictionary);
-
- byte[] recordBuffer = new byte[recordInfo.getByteFormLen()];
- final byte[] buffer = new byte[CoprocessorConstants.METRIC_SERIALIZE_BUFFER_SIZE];
-
- int iteratedSliceCount = 0;
- long latestSliceTs = Long.MIN_VALUE;
- for (Slice slice : slices) {
- latestSliceTs = slice.getTimestamp();
- iteratedSliceCount++;
-
- //dictionaries for fact table columns can not be determined while streaming.
- //a piece of dict coincide with each Slice, we call it "local dict"
- final Dictionary<?>[] localDictionaries = slice.getLocalDictionaries();
- CoprocessorFilter newFilter;
- final boolean emptyDictionary = Array.isEmpty(localDictionaries);
- if (emptyDictionary) {
- newFilter = filter;
- } else {
- for (Dictionary<?> localDictionary : localDictionaries) {
- if (localDictionary instanceof TrieDictionary) {
- ((TrieDictionary) localDictionary).enableIdToValueBytesCache();
- }
- }
- newFilter = CoprocessorFilter.fromFilter(new LocalDictionary(localDictionaries, type, slice.getInfo()), filter.getFilter(), FilterDecorator.FilterConstantsTreatment.REPLACE_WITH_LOCAL_DICT);
- }
-
- ConciseSet result = null;
- if (filter != null) {
- result = new BitMapFilterEvaluator(new SliceBitMapProvider(slice, type)).evaluate(newFilter.getFilter());
- }
-
- Iterator<RawTableRecord> iterator = slice.iterateWithBitmap(result);
-
- TblColRef[] columns = type.columns;
- int[] finalColumnLength = new int[columns.length];
- for (int i = 0; i < columns.length; ++i) {
- finalColumnLength[i] = rowKeyColumnIO.getColumnLength(columns[i]);
- }
-
- while (iterator.hasNext()) {
- final RawTableRecord rawTableRecord = iterator.next();
- decodeWithDictionary(recordBuffer, rawTableRecord, localDictionaries, recordInfo, rowKeyColumnIO, finalColumnLength);
-
- if (needAgg) {
- //if has group by, group them first, and extract entries later
- AggrKey aggKey = projector.getAggrKey(recordBuffer);
- MeasureAggregator[] bufs = aggCache.getBuffer(aggKey);
- aggregators.aggregate(bufs, recordBuffer);
- aggCache.checkMemoryUsage();
- } else {
- //otherwise directly extract entry and put into response
- if (totalByteFormLen >= MEMORY_LIMIT) {
- throw new RuntimeException("the query has exceeded the memory limit, please check the query");
- }
- IIProtos.IIResponseInternal.IIRow.Builder rowBuilder = IIProtos.IIResponseInternal.IIRow.newBuilder().setColumns(ByteString.copyFrom(recordBuffer));
- responseBuilder.addRows(rowBuilder.build());
- totalByteFormLen += byteFormLen;
- }
- }
- }
-
- logger.info("Iterated Slices count: " + iteratedSliceCount);
-
- if (needAgg) {
- for (Map.Entry<AggrKey, MeasureAggregator[]> entry : aggCache.getAllEntries()) {
- AggrKey aggrKey = entry.getKey();
- IIProtos.IIResponseInternal.IIRow.Builder rowBuilder = IIProtos.IIResponseInternal.IIRow.newBuilder().setColumns(ByteString.copyFrom(aggrKey.get(), aggrKey.offset(), aggrKey.length()));
- int length = aggregators.serializeMetricValues(entry.getValue(), buffer);
- rowBuilder.setMeasures(ByteString.copyFrom(buffer, 0, length));
- responseBuilder.addRows(rowBuilder.build());
- }
- }
-
- responseBuilder.setStats(IIProtos.IIResponseInternal.Stats.newBuilder().setLatestDataTime(latestSliceTs).setServiceStartTime(this.serviceStartTime).setServiceEndTime(System.currentTimeMillis()).setScannedSlices(iteratedSliceCount));
- return responseBuilder.build();
- }
-
- private void decodeWithDictionary(byte[] recordBuffer, RawTableRecord encodedRecord, Dictionary<?>[] localDictionaries, TableRecordInfoDigest digest, RowKeyColumnIO rowKeyColumnIO, int[] finalColumnLengths) {
- final boolean[] isMetric = digest.isMetrics();
- final boolean emptyDictionary = Array.isEmpty(localDictionaries);
- for (int i = 0; i < finalColumnLengths.length; i++) {
- if (isMetric[i]) {
- rowKeyColumnIO.writeColumnWithoutDictionary(encodedRecord.getBytes(), encodedRecord.offset(i), encodedRecord.length(i), recordBuffer, digest.offset(i), finalColumnLengths[i]);
- } else {
- if (emptyDictionary) {
- rowKeyColumnIO.writeColumnWithoutDictionary(encodedRecord.getBytes(), encodedRecord.offset(i), encodedRecord.length(i), recordBuffer, digest.offset(i), finalColumnLengths[i]);
- } else {
- final Dictionary<?> localDictionary = localDictionaries[i];
- final byte[] valueBytesFromId = localDictionary.getValueBytesFromId(encodedRecord.getValueID(i));
- rowKeyColumnIO.writeColumnWithoutDictionary(valueBytesFromId, 0, valueBytesFromId.length, recordBuffer, digest.offset(i), finalColumnLengths[i]);
- }
- }
- }
- }
-
- @Override
- public void start(CoprocessorEnvironment env) throws IOException {
- if (env instanceof RegionCoprocessorEnvironment) {
- this.env = (RegionCoprocessorEnvironment) env;
- } else {
- throw new CoprocessorException("Must be loaded on a table region!");
- }
- }
-
- @Override
- public void stop(CoprocessorEnvironment env) throws IOException {
- }
-
- @Override
- public Service getService() {
- return this;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/3d2685d8/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIResponseAdapter.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIResponseAdapter.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIResponseAdapter.java
deleted file mode 100644
index 8675de1..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIResponseAdapter.java
+++ /dev/null
@@ -1,24 +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.coprocessor.endpoint;
-
-/**
- */
-public class IIResponseAdapter {
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/3d2685d8/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/LocalDictionary.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/LocalDictionary.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/LocalDictionary.java
deleted file mode 100644
index 0fc0e08..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/LocalDictionary.java
+++ /dev/null
@@ -1,32 +0,0 @@
-package org.apache.kylin.storage.hbase.coprocessor.endpoint;
-
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.dict.IDictionaryAware;
-import org.apache.kylin.invertedindex.index.TableRecordInfoDigest;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorRowType;
-
-/**
- */
-public class LocalDictionary implements IDictionaryAware {
-
- private CoprocessorRowType type;
- private Dictionary<?>[] colDictMap;
- private TableRecordInfoDigest recordInfo;
-
- public LocalDictionary(Dictionary<?>[] colDictMap, CoprocessorRowType type, TableRecordInfoDigest recordInfo) {
- this.colDictMap = colDictMap;
- this.type = type;
- this.recordInfo = recordInfo;
- }
-
- @Override
- public int getColumnLength(TblColRef col) {
- return recordInfo.length(type.getColIndexByTblColRef(col));
- }
-
- @Override
- public Dictionary<?> getDictionary(TblColRef col) {
- return this.colDictMap[type.getColIndexByTblColRef(col)];
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/3d2685d8/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/SliceBitMapProvider.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/SliceBitMapProvider.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/SliceBitMapProvider.java
deleted file mode 100644
index 130a015..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/SliceBitMapProvider.java
+++ /dev/null
@@ -1,56 +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.coprocessor.endpoint;
-
-import org.apache.kylin.invertedindex.index.Slice;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.storage.filter.BitMapFilterEvaluator;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorRowType;
-
-import it.uniroma3.mat.extendedset.intset.ConciseSet;
-
-/**
- * <p/>
- * an adapter
- */
-public class SliceBitMapProvider implements BitMapFilterEvaluator.BitMapProvider {
-
- private Slice slice;
- private CoprocessorRowType type;
-
- public SliceBitMapProvider(Slice slice, CoprocessorRowType type) {
- this.slice = slice;
- this.type = type;
- }
-
- @Override
- public ConciseSet getBitMap(TblColRef col, Integer startId, Integer endId) {
- return slice.getColumnValueContainer(type.getColIndexByTblColRef(col)).getBitMap(startId, endId);
- }
-
- @Override
- public int getRecordCount() {
- return this.slice.getRecordCount();
- }
-
- @Override
- public int getMaxValueId(TblColRef col) {
- return slice.getColumnValueContainer(type.getColIndexByTblColRef(col)).getMaxValueId();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/3d2685d8/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java
deleted file mode 100644
index 27a919c..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java
+++ /dev/null
@@ -1,86 +0,0 @@
-package org.apache.kylin.storage.hbase.coprocessor.endpoint;
-
-import org.apache.kylin.common.util.DateFormat;
-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.model.TblColRef;
-
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
-
-/**
- */
-public class TsConditionExtractor {
-
- /**
- *
- * @return null if the ts range conflicts with each other
- * Ranges.all() if no ts condition is defined(in surfaced ANDs)
- */
- public static Range<Long> extractTsCondition(TblColRef tsColRef, TupleFilter rootFilter) {
- return extractTsConditionInternal(rootFilter, tsColRef);
- }
-
- private static Range<Long> extractTsConditionInternal(TupleFilter filter, TblColRef colRef) {
- if (filter == null) {
- return Ranges.all();
- }
-
- if (filter instanceof LogicalTupleFilter) {
- if (filter.getOperator() == TupleFilter.FilterOperatorEnum.AND) {
- Range<Long> ret = Ranges.all();
- for (TupleFilter child : filter.getChildren()) {
- Range childRange = extractTsConditionInternal(child, colRef);
- if (childRange != null) {
- if (ret.isConnected(childRange) && !ret.intersection(childRange).isEmpty()) {
- ret = ret.intersection(childRange);
- } else {
- return null;
- }
- } else {
- return null;
- }
- }
- return ret.isEmpty() ? null : ret;
- } else {
- //for conditions like date > DATE'2000-11-11' OR date < DATE '1999-01-01'
- //we will use Ranges.all() rather than two ranges to represent them
- return Ranges.all();
- }
- }
-
- if (filter instanceof CompareTupleFilter) {
- CompareTupleFilter compareTupleFilter = (CompareTupleFilter) filter;
- if (compareTupleFilter.getColumn() == null)// column will be null at filters like " 1<>1"
- return Ranges.all();
-
- if (compareTupleFilter.getColumn().equals(colRef)) {
- Object firstValue = compareTupleFilter.getFirstValue();
- long t;
- switch (compareTupleFilter.getOperator()) {
- case EQ:
- t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.closed(t, t);
- case LT:
- t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.lessThan(t);
- case LTE:
- t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.atMost(t);
- case GT:
- t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.greaterThan(t);
- case GTE:
- t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.atLeast(t);
- case NEQ:
- case IN://not handled for now
- break;
- default:
- }
- }
- }
- return Ranges.all();
- }
-}