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:35 UTC
[12/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/hbase/coprocessor/endpoint/protobuf/CubeVisit.proto
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/CubeVisit.proto b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/CubeVisit.proto
deleted file mode 100644
index fa7ab20..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/CubeVisit.proto
+++ /dev/null
@@ -1,33 +0,0 @@
-// usage:
-// protoc --java_out=./storage/src/main/java ./storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/CubeVisit.proto
-
-option java_package = "org.apache.kylin.storage.hbase.coprocessor.endpoint.generated";
-
-option java_outer_classname = "CubeVisitProtos";
-
-option java_generic_services = true;
-
-option java_generate_equals_and_hash = true;
-
-option optimize_for = SPEED;
-
-message CubeVisitRequest {
- required bytes gtScanRequest = 1;
- required bytes hbaseRawScan = 2;
-}
-
-message CubeVisitResponse {
- //all entries in this struct be optional to conveniently add more entries in the future
- message Stats {
- optional int64 serviceStartTime = 1;
- optional int64 serviceEndTime = 2;
- optional int32 scannedRowCount = 3;
- optional int32 aggregatedRowCount = 4;
- }
- required bytes compressedRows = 1;
- required Stats stats = 2;
-}
-
-service CubeVisitService {
- rpc visitCube (CubeVisitRequest) returns (CubeVisitResponse);
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/II.proto
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/II.proto b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/II.proto
deleted file mode 100644
index de7081f..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/II.proto
+++ /dev/null
@@ -1,47 +0,0 @@
-// usage:
-// protoc --java_out=./storage/src/main/java ./storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/protobuf/II.proto
-
-option java_package = "org.apache.kylin.storage.hbase.coprocessor.endpoint.generated";
-
-option java_outer_classname = "IIProtos";
-
-option java_generic_services = true;
-
-option java_generate_equals_and_hash = true;
-
-option optimize_for = SPEED;
-
-message IIRequest {
- required bytes type = 1;
- required bytes filter = 2;
- required bytes projector = 3;
- required bytes aggregator = 4;
- optional bytes tsRange = 5;
-}
-
-message IIResponseInternal {
- message IIRow {
- required bytes columns = 1;
- optional bytes measures = 2;
- }
- //all entries in this struct be optional to conveniently add more entries in the future
- message Stats {
- optional int32 myShard = 1;
- optional int64 latestDataTime = 2;
- optional int64 serviceStartTime = 3;
- optional int64 serviceEndTime = 4;
- optional int32 scannedSlices = 5;
- }
- repeated IIRow rows = 1;
- required Stats stats = 2;
-}
-
-message IIResponse {
- required bytes blob = 1;
-}
-
-
-
-service RowsService {
- rpc getRows (IIRequest) returns (IIResponse);
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregateRegionObserver.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregateRegionObserver.java
deleted file mode 100644
index 51dc81c..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregateRegionObserver.java
+++ /dev/null
@@ -1,111 +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.observer;
-
-import java.io.IOException;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
-import org.apache.hadoop.hbase.coprocessor.ObserverContext;
-import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.hadoop.hbase.regionserver.HRegion;
-import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorFilter;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorProjector;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorRowType;
-
-/**
- * @author yangli9
- */
-public class AggregateRegionObserver extends BaseRegionObserver {
-
- // HBase uses common logging (vs. Kylin uses slf4j)
- static final Log LOG = LogFactory.getLog(AggregateRegionObserver.class);
-
- static final String COPROCESSOR_ENABLE = "_Coprocessor_Enable";
- static final String TYPE = "_Type";
- static final String PROJECTOR = "_Projector";
- static final String AGGREGATORS = "_Aggregators";
- static final String FILTER = "_Filter";
- static final String BEHAVIOR = "_Behavior";
-
- @Override
- public final RegionScanner postScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> ctxt, final Scan scan, final RegionScanner innerScanner) throws IOException {
-
- boolean copAbortOnError = ctxt.getEnvironment().getConfiguration().getBoolean(RegionCoprocessorHost.ABORT_ON_ERROR_KEY, RegionCoprocessorHost.DEFAULT_ABORT_ON_ERROR);
-
- // never throw out exception that could abort region server
- if (copAbortOnError) {
- try {
- return doPostScannerObserver(ctxt, scan, innerScanner);
- } catch (Throwable e) {
- LOG.error("Kylin Coprocessor Error", e);
- return innerScanner;
- }
- } else {
- return doPostScannerObserver(ctxt, scan, innerScanner);
- }
- }
-
- private RegionScanner doPostScannerObserver(final ObserverContext<RegionCoprocessorEnvironment> ctxt, final Scan scan, final RegionScanner innerScanner) throws IOException {
- byte[] coprocessorEnableBytes = scan.getAttribute(COPROCESSOR_ENABLE);
- if (coprocessorEnableBytes == null || coprocessorEnableBytes.length == 0 || coprocessorEnableBytes[0] == 0) {
- return innerScanner;
- }
-
- byte[] typeBytes = scan.getAttribute(TYPE);
- CoprocessorRowType type = CoprocessorRowType.deserialize(typeBytes);
-
- byte[] projectorBytes = scan.getAttribute(PROJECTOR);
- CoprocessorProjector projector = CoprocessorProjector.deserialize(projectorBytes);
-
- byte[] aggregatorBytes = scan.getAttribute(AGGREGATORS);
- ObserverAggregators aggregators = ObserverAggregators.deserialize(aggregatorBytes);
-
- byte[] filterBytes = scan.getAttribute(FILTER);
- CoprocessorFilter filter = CoprocessorFilter.deserialize(filterBytes);
-
- ObserverBehavior observerBehavior = ObserverBehavior.SCAN_FILTER_AGGR_CHECKMEM;
- try {
- byte[] behavior = scan.getAttribute(BEHAVIOR);
- if (behavior != null && behavior.length != 0) {
- observerBehavior = ObserverBehavior.valueOf(new String(behavior));
- }
- } catch (Exception e) {
- LOG.error("failed to parse behavior,using default behavior SCAN_FILTER_AGGR_CHECKMEM", e);
- observerBehavior = ObserverBehavior.SCAN_FILTER_AGGR_CHECKMEM;
- }
-
- // start/end region operation & sync on scanner is suggested by the
- // javadoc of RegionScanner.nextRaw()
- // FIXME: will the lock still work when a iterator is returned? is it safe? Is readonly attribute helping here? by mhb
- HRegion region = ctxt.getEnvironment().getRegion();
- region.startRegionOperation();
- try {
- synchronized (innerScanner) {
- return new AggregationScanner(type, filter, projector, aggregators, innerScanner, observerBehavior);
- }
- } finally {
- region.closeRegionOperation();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregationScanner.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregationScanner.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregationScanner.java
deleted file mode 100644
index a8d5d99..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/AggregationScanner.java
+++ /dev/null
@@ -1,172 +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.observer;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.HRegionInfo;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.kylin.metadata.measure.MeasureAggregator;
-import org.apache.kylin.storage.hbase.coprocessor.AggrKey;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorFilter;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorProjector;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorRowType;
-
-/**
- * @author yangli9
- *
- */
-public class AggregationScanner implements RegionScanner {
-
- private RegionScanner outerScanner;
- private ObserverBehavior behavior;
-
- public AggregationScanner(CoprocessorRowType type, CoprocessorFilter filter, CoprocessorProjector groupBy, ObserverAggregators aggrs, RegionScanner innerScanner, ObserverBehavior behavior) throws IOException {
-
- AggregateRegionObserver.LOG.info("Kylin Coprocessor start");
-
- this.behavior = behavior;
-
- ObserverAggregationCache aggCache;
- Stats stats = new Stats();
-
- aggCache = buildAggrCache(innerScanner, type, groupBy, aggrs, filter, stats);
- stats.countOutputRow(aggCache.getSize());
- this.outerScanner = aggCache.getScanner(innerScanner);
-
- AggregateRegionObserver.LOG.info("Kylin Coprocessor aggregation done: " + stats);
- }
-
- @SuppressWarnings("rawtypes")
- ObserverAggregationCache buildAggrCache(final RegionScanner innerScanner, CoprocessorRowType type, CoprocessorProjector projector, ObserverAggregators aggregators, CoprocessorFilter filter, Stats stats) throws IOException {
-
- ObserverAggregationCache aggCache = new ObserverAggregationCache(aggregators);
-
- ObserverTuple tuple = new ObserverTuple(type);
- boolean hasMore = true;
- List<Cell> results = new ArrayList<Cell>();
- while (hasMore) {
- results.clear();
- hasMore = innerScanner.nextRaw(results);
- if (results.isEmpty())
- continue;
-
- if (stats != null)
- stats.countInputRow(results);
-
- Cell cell = results.get(0);
- tuple.setUnderlying(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
-
- if (behavior.ordinal() >= ObserverBehavior.SCAN_FILTER.ordinal()) {
- if (filter != null && filter.evaluate(tuple) == false)
- continue;
-
- if (behavior.ordinal() >= ObserverBehavior.SCAN_FILTER_AGGR.ordinal()) {
- AggrKey aggKey = projector.getAggrKey(results);
- MeasureAggregator[] bufs = aggCache.getBuffer(aggKey);
- aggregators.aggregate(bufs, results);
-
- if (behavior.ordinal() >= ObserverBehavior.SCAN_FILTER_AGGR_CHECKMEM.ordinal()) {
- aggCache.checkMemoryUsage();
- }
- }
- }
- }
- return aggCache;
- }
-
- @Override
- public boolean next(List<Cell> results) throws IOException {
- return outerScanner.next(results);
- }
-
- @Override
- public boolean next(List<Cell> result, int limit) throws IOException {
- return outerScanner.next(result, limit);
- }
-
- @Override
- public boolean nextRaw(List<Cell> result) throws IOException {
- return outerScanner.nextRaw(result);
- }
-
- @Override
- public boolean nextRaw(List<Cell> result, int limit) throws IOException {
- return outerScanner.nextRaw(result, limit);
- }
-
- @Override
- public void close() throws IOException {
- outerScanner.close();
- }
-
- @Override
- public HRegionInfo getRegionInfo() {
- return outerScanner.getRegionInfo();
- }
-
- @Override
- public boolean isFilterDone() throws IOException {
- return outerScanner.isFilterDone();
- }
-
- @Override
- public boolean reseek(byte[] row) throws IOException {
- return outerScanner.reseek(row);
- }
-
- @Override
- public long getMaxResultSize() {
- return outerScanner.getMaxResultSize();
- }
-
- @Override
- public long getMvccReadPoint() {
- return outerScanner.getMvccReadPoint();
- }
-
- private static class Stats {
- long inputRows = 0;
- long inputBytes = 0;
- long outputRows = 0;
-
- // have no outputBytes because that requires actual serialize all the
- // aggregator buffers
-
- public void countInputRow(List<Cell> row) {
- inputRows++;
- inputBytes += row.get(0).getRowLength();
- for (int i = 0, n = row.size(); i < n; i++) {
- inputBytes += row.get(i).getValueLength();
- }
- }
-
- public void countOutputRow(long rowCount) {
- outputRows += rowCount;
- }
-
- public String toString() {
- double percent = (double) outputRows / inputRows * 100;
- return Math.round(percent) + "% = " + outputRows + " (out rows) / " + inputRows + " (in rows); in bytes = " + inputBytes + "; est. out bytes = " + Math.round(inputBytes * percent / 100);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregationCache.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregationCache.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregationCache.java
deleted file mode 100644
index 543ae8c..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregationCache.java
+++ /dev/null
@@ -1,161 +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.observer;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map.Entry;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.HRegionInfo;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.KeyValue.Type;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.kylin.metadata.measure.MeasureAggregator;
-import org.apache.kylin.storage.hbase.coprocessor.AggrKey;
-import org.apache.kylin.storage.hbase.coprocessor.AggregationCache;
-
-/**
- * @author yangli9
- */
-@SuppressWarnings("rawtypes")
-public class ObserverAggregationCache extends AggregationCache {
-
- private final ObserverAggregators aggregators;
-
- public ObserverAggregationCache(ObserverAggregators aggregators) {
- this.aggregators = aggregators;
- }
-
- public RegionScanner getScanner(RegionScanner innerScanner) {
- return new AggregationRegionScanner(innerScanner);
- }
-
- @Override
- public MeasureAggregator[] createBuffer() {
- return aggregators.createBuffer();
- }
-
- private class AggregationRegionScanner implements RegionScanner {
-
- private final RegionScanner innerScanner;
- private final Iterator<Entry<AggrKey, MeasureAggregator[]>> iterator;
-
- public AggregationRegionScanner(RegionScanner innerScanner) {
- this.innerScanner = innerScanner;
- this.iterator = aggBufMap.entrySet().iterator();
- }
-
- @Override
- public boolean next(List<Cell> results) throws IOException {
- // AggregateRegionObserver.LOG.info("Kylin Scanner next()");
- boolean hasMore = false;
- if (iterator.hasNext()) {
- Entry<AggrKey, MeasureAggregator[]> entry = iterator.next();
- makeCells(entry, results);
- hasMore = iterator.hasNext();
- }
- // AggregateRegionObserver.LOG.info("Kylin Scanner next() done");
- return hasMore;
- }
-
- private void makeCells(Entry<AggrKey, MeasureAggregator[]> entry, List<Cell> results) {
- byte[][] families = aggregators.getHColFamilies();
- byte[][] qualifiers = aggregators.getHColQualifiers();
- int nHCols = aggregators.getHColsNum();
-
- AggrKey rowKey = entry.getKey();
- MeasureAggregator[] aggBuf = entry.getValue();
- ByteBuffer[] rowValues = aggregators.getHColValues(aggBuf);
-
- if (nHCols == 0) {
- Cell keyValue = new KeyValue(rowKey.get(), rowKey.offset(), rowKey.length(), //
- null, 0, 0, //
- null, 0, 0, //
- HConstants.LATEST_TIMESTAMP, Type.Put, //
- null, 0, 0);
- results.add(keyValue);
- } else {
- for (int i = 0; i < nHCols; i++) {
- Cell keyValue = new KeyValue(rowKey.get(), rowKey.offset(), rowKey.length(), //
- families[i], 0, families[i].length, //
- qualifiers[i], 0, qualifiers[i].length, //
- HConstants.LATEST_TIMESTAMP, Type.Put, //
- rowValues[i].array(), 0, rowValues[i].position());
- results.add(keyValue);
- }
- }
- }
-
- @Override
- public boolean next(List<Cell> result, int limit) throws IOException {
- return next(result);
- }
-
- @Override
- public boolean nextRaw(List<Cell> result) throws IOException {
- return next(result);
- }
-
- @Override
- public boolean nextRaw(List<Cell> result, int limit) throws IOException {
- return next(result);
- }
-
- @Override
- public void close() throws IOException {
- // AggregateRegionObserver.LOG.info("Kylin Scanner close()");
- innerScanner.close();
- // AggregateRegionObserver.LOG.info("Kylin Scanner close() done");
- }
-
- @Override
- public HRegionInfo getRegionInfo() {
- // AggregateRegionObserver.LOG.info("Kylin Scanner getRegionInfo()");
- return innerScanner.getRegionInfo();
- }
-
- @Override
- public long getMaxResultSize() {
- // AggregateRegionObserver.LOG.info("Kylin Scanner getMaxResultSize()");
- return Long.MAX_VALUE;
- }
-
- @Override
- public boolean isFilterDone() throws IOException {
- // AggregateRegionObserver.LOG.info("Kylin Scanner isFilterDone()");
- return false;
- }
-
- @Override
- public boolean reseek(byte[] row) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public long getMvccReadPoint() {
- // AggregateRegionObserver.LOG.info("Kylin Scanner getMvccReadPoint()");
- return Long.MAX_VALUE;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregators.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregators.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregators.java
deleted file mode 100644
index bc1c362..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverAggregators.java
+++ /dev/null
@@ -1,261 +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.observer;
-
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.List;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.common.util.BytesSerializer;
-import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.cube.model.HBaseColumnDesc;
-import org.apache.kylin.metadata.measure.MeasureAggregator;
-import org.apache.kylin.metadata.measure.MeasureCodec;
-import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorConstants;
-import org.apache.kylin.storage.hbase.steps.RowValueDecoder;
-
-/**
- * @author yangli9
- */
-@SuppressWarnings({ "rawtypes", "unchecked" })
-public class ObserverAggregators {
-
- public static ObserverAggregators fromValueDecoders(Collection<RowValueDecoder> rowValueDecoders) {
-
- // each decoder represents one HBase column
- HCol[] hcols = new HCol[rowValueDecoders.size()];
- int i = 0;
- for (RowValueDecoder rowValueDecoder : rowValueDecoders) {
- hcols[i++] = buildHCol(rowValueDecoder.getHBaseColumn());
- }
-
- ObserverAggregators aggrs = new ObserverAggregators(hcols);
- return aggrs;
-
- }
-
- private static HCol buildHCol(HBaseColumnDesc desc) {
- byte[] family = Bytes.toBytes(desc.getColumnFamilyName());
- byte[] qualifier = Bytes.toBytes(desc.getQualifier());
- MeasureDesc[] measures = desc.getMeasures();
-
- String[] funcNames = new String[measures.length];
- String[] dataTypes = new String[measures.length];
-
- for (int i = 0; i < measures.length; i++) {
- funcNames[i] = measures[i].getFunction().getExpression();
- dataTypes[i] = measures[i].getFunction().getReturnType();
- }
-
- return new HCol(family, qualifier, funcNames, dataTypes);
- }
-
- public static byte[] serialize(ObserverAggregators o) {
- ByteBuffer buf = ByteBuffer.allocate(CoprocessorConstants.SERIALIZE_BUFFER_SIZE);
- serializer.serialize(o, buf);
- byte[] result = new byte[buf.position()];
- System.arraycopy(buf.array(), 0, result, 0, buf.position());
- return result;
- }
-
- public static ObserverAggregators deserialize(byte[] bytes) {
- return serializer.deserialize(ByteBuffer.wrap(bytes));
- }
-
- private static final Serializer serializer = new Serializer();
-
- private static class Serializer implements BytesSerializer<ObserverAggregators> {
-
- @Override
- public void serialize(ObserverAggregators value, ByteBuffer out) {
- BytesUtil.writeVInt(value.nHCols, out);
- for (int i = 0; i < value.nHCols; i++) {
- HCol col = value.hcols[i];
- BytesUtil.writeByteArray(col.family, out);
- BytesUtil.writeByteArray(col.qualifier, out);
- BytesUtil.writeAsciiStringArray(col.funcNames, out);
- BytesUtil.writeAsciiStringArray(col.dataTypes, out);
- }
- }
-
- @Override
- public ObserverAggregators deserialize(ByteBuffer in) {
- int nHCols = BytesUtil.readVInt(in);
- HCol[] hcols = new HCol[nHCols];
- for (int i = 0; i < nHCols; i++) {
- byte[] family = BytesUtil.readByteArray(in);
- byte[] qualifier = BytesUtil.readByteArray(in);
- String[] funcNames = BytesUtil.readAsciiStringArray(in);
- String[] dataTypes = BytesUtil.readAsciiStringArray(in);
- hcols[i] = new HCol(family, qualifier, funcNames, dataTypes);
- }
- return new ObserverAggregators(hcols);
- }
-
- }
-
- // ============================================================================
-
- final HCol[] hcols;
- final int nHCols;
- final ByteBuffer[] hColValues;
- final int nTotalMeasures;
-
- public ObserverAggregators(HCol[] _hcols) {
- this.hcols = sort(_hcols);
- this.nHCols = hcols.length;
- this.hColValues = new ByteBuffer[nHCols];
-
- int nTotalMeasures = 0;
- for (HCol col : hcols)
- nTotalMeasures += col.nMeasures;
- this.nTotalMeasures = nTotalMeasures;
- }
-
- private HCol[] sort(HCol[] hcols) {
- HCol[] copy = Arrays.copyOf(hcols, hcols.length);
- Arrays.sort(copy, new Comparator<HCol>() {
- @Override
- public int compare(HCol o1, HCol o2) {
- int comp = Bytes.compareTo(o1.family, o2.family);
- if (comp != 0)
- return comp;
- comp = Bytes.compareTo(o1.qualifier, o2.qualifier);
- return comp;
- }
- });
- return copy;
- }
-
- public MeasureAggregator[] createBuffer() {
- MeasureAggregator[] aggrs = new MeasureAggregator[nTotalMeasures];
- int i = 0;
- for (HCol col : hcols) {
- for (int j = 0; j < col.nMeasures; j++)
- aggrs[i++] = MeasureAggregator.create(col.funcNames[j], col.dataTypes[j]);
- }
- return aggrs;
- }
-
- public void aggregate(MeasureAggregator[] measureAggrs, List<Cell> rowCells) {
- int i = 0;
- for (int ci = 0; ci < nHCols; ci++) {
- HCol col = hcols[ci];
- Cell cell = findCell(col, rowCells);
-
- if (cell == null) {
- i += col.nMeasures;
- continue;
- }
-
- ByteBuffer input = ByteBuffer.wrap(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
-
- col.measureCodec.decode(input, col.measureValues);
- for (int j = 0; j < col.nMeasures; j++)
- measureAggrs[i++].aggregate(col.measureValues[j]);
- }
- }
-
- private Cell findCell(HCol col, List<Cell> cells) {
- // cells are ordered by timestamp asc, thus search from back, first hit
- // is the latest version
- for (int i = cells.size() - 1; i >= 0; i--) {
- Cell cell = cells.get(i);
- if (match(col, cell)) {
- return cell;
- }
- }
- return null;
- }
-
- public static boolean match(HCol col, Cell cell) {
- return Bytes.compareTo(col.family, 0, col.family.length, cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()) == 0 && Bytes.compareTo(col.qualifier, 0, col.qualifier.length, cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()) == 0;
- }
-
- public int getHColsNum() {
- return nHCols;
- }
-
- public byte[][] getHColFamilies() {
- byte[][] result = new byte[nHCols][];
- for (int i = 0; i < nHCols; i++)
- result[i] = hcols[i].family;
- return result;
- }
-
- public byte[][] getHColQualifiers() {
- byte[][] result = new byte[nHCols][];
- for (int i = 0; i < nHCols; i++)
- result[i] = hcols[i].qualifier;
- return result;
- }
-
- public ByteBuffer[] getHColValues(MeasureAggregator[] aggrs) {
- int i = 0;
- for (int ci = 0; ci < nHCols; ci++) {
- HCol col = hcols[ci];
- for (int j = 0; j < col.nMeasures; j++)
- col.measureValues[j] = aggrs[i++].getState();
-
- col.measureBuf.clear();
- col.measureCodec.encode(col.measureValues, col.measureBuf);
- hColValues[ci] = col.measureBuf;
- }
- return hColValues;
- }
-
- // ============================================================================
-
- public static class HCol {
- final byte[] family;
- final byte[] qualifier;
- final String[] funcNames;
- final String[] dataTypes;
- final int nMeasures;
-
- final MeasureCodec measureCodec;
- final Object[] measureValues;
- final ByteBuffer measureBuf;
-
- public HCol(byte[] bFamily, byte[] bQualifier, String[] funcNames, String[] dataTypes) {
- this.family = bFamily;
- this.qualifier = bQualifier;
- this.funcNames = funcNames;
- this.dataTypes = dataTypes;
- this.nMeasures = funcNames.length;
- assert funcNames.length == dataTypes.length;
-
- this.measureCodec = new MeasureCodec(dataTypes);
- this.measureValues = new Object[nMeasures];
- this.measureBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
- }
-
- @Override
- public String toString() {
- return "HCol [bFamily=" + Bytes.toString(family) + ", bQualifier=" + Bytes.toString(qualifier) + ", nMeasures=" + nMeasures + "]";
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverBehavior.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverBehavior.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverBehavior.java
deleted file mode 100644
index ff6060d..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverBehavior.java
+++ /dev/null
@@ -1,10 +0,0 @@
-package org.apache.kylin.storage.hbase.coprocessor.observer;
-
-/**
- */
-public enum ObserverBehavior {
- SCAN, //only scan data, used for profiling tuple scan speed
- SCAN_FILTER, //only scan+filter used,used for profiling filter speed
- SCAN_FILTER_AGGR, //aggregate the result
- SCAN_FILTER_AGGR_CHECKMEM, //default full operations
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverEnabler.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverEnabler.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverEnabler.java
deleted file mode 100644
index 90d701f..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverEnabler.java
+++ /dev/null
@@ -1,186 +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.observer;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.debug.BackdoorToggles;
-import org.apache.kylin.cube.CubeInstance;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.cuboid.Cuboid;
-import org.apache.kylin.metadata.filter.TupleFilter;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.storage.StorageContext;
-import org.apache.kylin.storage.hbase.RegionScannerAdapter;
-import org.apache.kylin.storage.hbase.ResultScannerAdapter;
-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.steps.RowValueDecoder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-
-/**
- * @author yangli9
- */
-public class ObserverEnabler {
-
- private static final Logger logger = LoggerFactory.getLogger(ObserverEnabler.class);
-
- static final String FORCE_COPROCESSOR = "forceObserver";
- static final Map<String, Boolean> CUBE_OVERRIDES = Maps.newConcurrentMap();
-
- public static ResultScanner scanWithCoprocessorIfBeneficial(CubeSegment segment, Cuboid cuboid, TupleFilter tupleFiler, //
- Collection<TblColRef> groupBy, Collection<RowValueDecoder> rowValueDecoders, StorageContext context, HTableInterface table, Scan scan) throws IOException {
-
- if (context.isCoprocessorEnabled() == false) {
- return table.getScanner(scan);
- }
-
- CoprocessorRowType type = CoprocessorRowType.fromCuboid(segment, cuboid);
- CoprocessorFilter filter = CoprocessorFilter.fromFilter(segment, tupleFiler, FilterDecorator.FilterConstantsTreatment.REPLACE_WITH_GLOBAL_DICT);
- CoprocessorProjector projector = CoprocessorProjector.makeForObserver(segment, cuboid, groupBy);
- ObserverAggregators aggrs = ObserverAggregators.fromValueDecoders(rowValueDecoders);
-
- boolean localCoprocessor = KylinConfig.getInstanceFromEnv().getQueryRunLocalCoprocessor() || BackdoorToggles.getRunLocalCoprocessor();
-
- if (localCoprocessor) {
- RegionScanner innerScanner = new RegionScannerAdapter(table.getScanner(scan));
- AggregationScanner aggrScanner = new AggregationScanner(type, filter, projector, aggrs, innerScanner, ObserverBehavior.SCAN_FILTER_AGGR_CHECKMEM);
- return new ResultScannerAdapter(aggrScanner);
- } else {
-
- // debug/profiling purpose
- String toggle = BackdoorToggles.getObserverBehavior();
- if (toggle == null) {
- toggle = ObserverBehavior.SCAN_FILTER_AGGR_CHECKMEM.toString(); //default behavior
- } else {
- logger.info("The execution of this query will use " + toggle + " as observer's behavior");
- }
-
- scan.setAttribute(AggregateRegionObserver.COPROCESSOR_ENABLE, new byte[] { 0x01 });
- scan.setAttribute(AggregateRegionObserver.BEHAVIOR, toggle.getBytes());
- scan.setAttribute(AggregateRegionObserver.TYPE, CoprocessorRowType.serialize(type));
- scan.setAttribute(AggregateRegionObserver.PROJECTOR, CoprocessorProjector.serialize(projector));
- scan.setAttribute(AggregateRegionObserver.AGGREGATORS, ObserverAggregators.serialize(aggrs));
- scan.setAttribute(AggregateRegionObserver.FILTER, CoprocessorFilter.serialize(filter));
- return table.getScanner(scan);
- }
- }
-
- public static void enableCoprocessorIfBeneficial(CubeInstance cube, Collection<TblColRef> groupBy, Collection<RowValueDecoder> rowValueDecoders, StorageContext context) {
- if (isCoprocessorBeneficial(cube, groupBy, rowValueDecoders, context)) {
- context.enableCoprocessor();
- }
- }
-
- private static boolean isCoprocessorBeneficial(CubeInstance cube, Collection<TblColRef> groupBy, Collection<RowValueDecoder> rowValueDecoders, StorageContext context) {
-
- String forceFlag = System.getProperty(FORCE_COPROCESSOR);
- if (forceFlag != null) {
- return Boolean.parseBoolean(forceFlag);
- }
-
- Boolean cubeOverride = CUBE_OVERRIDES.get(cube.getName());
- if (cubeOverride != null) {
- return cubeOverride.booleanValue();
- }
-
- // if (RowValueDecoder.hasMemHungryCountDistinct(rowValueDecoders)) {
- // logger.info("Coprocessor is disabled because there is memory hungry count distinct");
- // return false;
- // }
-
- if (context.isExactAggregation()) {
- logger.info("Coprocessor is disabled because exactAggregation is true");
- return false;
- }
-
- Cuboid cuboid = context.getCuboid();
- Set<TblColRef> toAggr = Sets.newHashSet(cuboid.getAggregationColumns());
- toAggr.removeAll(groupBy);
- if (toAggr.isEmpty()) {
- logger.info("Coprocessor is disabled because no additional columns to aggregate");
- return false;
- }
-
- logger.info("Coprocessor is enabled to aggregate " + toAggr + ", returning " + groupBy);
- return true;
- }
-
- @SuppressWarnings("unused")
- private static int getBitsToScan(byte[] startKey, byte[] stopKey) {
- // find the first bit difference from the beginning
- int totalBits = startKey.length * 8;
- int bitsToScan = totalBits;
- for (int i = 0; i < totalBits; i++) {
- int byteIdx = i / 8;
- int bitIdx = 7 - i % 8;
- byte bitMask = (byte) (1 << bitIdx);
- if ((startKey[byteIdx] & bitMask) == (stopKey[byteIdx] & bitMask))
- bitsToScan--;
- else
- break;
- }
- return bitsToScan;
- }
-
- public static void forceCoprocessorOn() {
- System.setProperty(FORCE_COPROCESSOR, "true");
- }
-
- public static void forceCoprocessorOff() {
- System.setProperty(FORCE_COPROCESSOR, "false");
- }
-
- public static String getForceCoprocessor() {
- return System.getProperty(FORCE_COPROCESSOR);
- }
-
- public static void forceCoprocessorUnset() {
- System.clearProperty(FORCE_COPROCESSOR);
- }
-
- public static void updateCubeOverride(String cubeName, String force) {
- if ("null".equalsIgnoreCase(force) || "default".equalsIgnoreCase(force)) {
- CUBE_OVERRIDES.remove(cubeName);
- } else if ("true".equalsIgnoreCase(force)) {
- CUBE_OVERRIDES.put(cubeName, Boolean.TRUE);
- } else if ("false".equalsIgnoreCase(force)) {
- CUBE_OVERRIDES.put(cubeName, Boolean.FALSE);
- }
- }
-
- public static Map<String, Boolean> getCubeOverrides() {
- return CUBE_OVERRIDES;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverTuple.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverTuple.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverTuple.java
deleted file mode 100644
index cdbe392..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/observer/ObserverTuple.java
+++ /dev/null
@@ -1,71 +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.observer;
-
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.tuple.IEvaluatableTuple;
-import org.apache.kylin.storage.hbase.coprocessor.CoprocessorRowType;
-
-/**
- * A special kind of tuple that exposes column value (dictionary ID) directly on
- * top of row key.
- *
- * @author yangli9
- */
-public class ObserverTuple implements IEvaluatableTuple {
-
- final CoprocessorRowType type;
-
- ImmutableBytesWritable rowkey;
- String[] values;
-
- public ObserverTuple(CoprocessorRowType type) {
- this.type = type;
- this.rowkey = new ImmutableBytesWritable();
- this.values = new String[type.getColumnCount()];
- }
-
- public void setUnderlying(byte[] array, int offset, int length) {
- rowkey.set(array, offset, length);
- for (int i = 0; i < values.length; i++) {
- values[i] = null;
- }
- }
-
- private String getValueAt(int i) {
- int n = type.getColumnCount();
- if (i < 0 || i >= n)
- return null;
-
- if (values[i] == null) {
- values[i] = Dictionary.dictIdToString(rowkey.get(), rowkey.getOffset() + type.columnOffsets[i], type.columnSizes[i]);
- }
-
- return values[i];
- }
-
- @Override
- public Object getValue(TblColRef col) {
- int i = type.getColIndexByTblColRef(col);
- return getValueAt(i);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridInstance.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridInstance.java b/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridInstance.java
deleted file mode 100644
index 997e7b2..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridInstance.java
+++ /dev/null
@@ -1,268 +0,0 @@
-package org.apache.kylin.storage.hybrid;
-
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.LinkedHashSet;
-import java.util.List;
-
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.persistence.ResourceStore;
-import org.apache.kylin.common.persistence.RootPersistentEntity;
-import org.apache.kylin.metadata.model.DataModelDesc;
-import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.project.RealizationEntry;
-import org.apache.kylin.metadata.realization.IRealization;
-import org.apache.kylin.metadata.realization.RealizationRegistry;
-import org.apache.kylin.metadata.realization.RealizationType;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.Lists;
-
-/**
- */
-
-@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
-public class HybridInstance extends RootPersistentEntity implements IRealization {
-
- @JsonIgnore
- private KylinConfig config;
-
- @JsonProperty("name")
- private String name;
-
- @JsonProperty("realizations")
- private List<RealizationEntry> realizationEntries;
-
- @JsonProperty("cost")
- private int cost = 50;
-
- private IRealization[] realizations = null;
- private List<TblColRef> allDimensions = null;
- private List<TblColRef> allColumns = null;
- private List<MeasureDesc> allMeasures = null;
- private long dateRangeStart;
- private long dateRangeEnd;
- private boolean isReady = false;
- private String projectName;
-
- private boolean initiated = false;
- private final static Logger logger = LoggerFactory.getLogger(HybridInstance.class);
-
- public List<RealizationEntry> getRealizationEntries() {
- return realizationEntries;
- }
-
- private void init() {
- if (initiated == true)
- return;
-
- synchronized (this) {
- if (initiated == true)
- return;
-
- if (realizationEntries == null || realizationEntries.size() == 0)
- throw new IllegalArgumentException();
-
- RealizationRegistry registry = RealizationRegistry.getInstance(config);
- List<IRealization> realizationList = Lists.newArrayList();
- for (int i = 0; i < realizationEntries.size(); i++) {
- IRealization realization = registry.getRealization(realizationEntries.get(i).getType(), realizationEntries.get(i).getRealization());
- if (realization == null) {
- logger.error("Realization '" + realizationEntries.get(i) + " is not found, remove from Hybrid '" + this.getName() + "'");
- continue;
- }
- if (realization.isReady() == false) {
- logger.error("Realization '" + realization.getName() + " is disabled, remove from Hybrid '" + this.getName() + "'");
- continue;
- }
- realizationList.add(realization);
- }
-
- LinkedHashSet<TblColRef> columns = new LinkedHashSet<TblColRef>();
- LinkedHashSet<TblColRef> dimensions = new LinkedHashSet<TblColRef>();
- LinkedHashSet<MeasureDesc> measures = new LinkedHashSet<MeasureDesc>();
- dateRangeStart = 0;
- dateRangeEnd = Long.MAX_VALUE;
- for (IRealization realization : realizationList) {
- columns.addAll(realization.getAllColumns());
- dimensions.addAll(realization.getAllDimensions());
- measures.addAll(realization.getMeasures());
-
- if (realization.isReady())
- isReady = true;
-
- if (dateRangeStart == 0 || realization.getDateRangeStart() < dateRangeStart)
- dateRangeStart = realization.getDateRangeStart();
-
- if (dateRangeStart == Long.MAX_VALUE || realization.getDateRangeEnd() > dateRangeEnd)
- dateRangeEnd = realization.getDateRangeEnd();
- }
-
- allDimensions = Lists.newArrayList(dimensions);
- allColumns = Lists.newArrayList(columns);
- allMeasures = Lists.newArrayList(measures);
-
- Collections.sort(realizationList, new Comparator<IRealization>() {
- @Override
- public int compare(IRealization o1, IRealization o2) {
-
- long i1 = o1.getDateRangeStart();
- long i2 = o2.getDateRangeStart();
- long comp = i1 - i2;
- if (comp != 0) {
- return comp > 0 ? 1 : -1;
- }
-
- i1 = o1.getDateRangeEnd();
- i2 = o2.getDateRangeEnd();
- comp = i1 - i2;
- if (comp != 0) {
- return comp > 0 ? 1 : -1;
- }
-
- return 0;
- }
- });
-
- this.realizations = realizationList.toArray(new IRealization[realizationList.size()]);
- initiated = true;
- }
- }
-
- @Override
- public boolean isCapable(SQLDigest digest) {
- for (IRealization realization : getRealizations()) {
- if (realization.isCapable(digest))
- return true;
- }
- return false;
- }
-
- @Override
- public int getCost(SQLDigest digest) {
- cost = 100;
- for (IRealization realization : this.getRealizations()) {
- if (realization.isCapable(digest))
- cost = Math.min(cost, realization.getCost(digest));
- }
-
- // Make hybrid always win its children
- return cost - 1;
- }
-
- @Override
- public RealizationType getType() {
- return RealizationType.HYBRID;
- }
-
- @Override
- public DataModelDesc getDataModelDesc() {
- return this.getLatestRealization().getDataModelDesc();
- }
-
- @Override
- public String getFactTable() {
- return getRealizations()[0].getFactTable();
- }
-
- @Override
- public List<TblColRef> getAllColumns() {
- init();
- return allColumns;
- }
-
- @Override
- public List<MeasureDesc> getMeasures() {
- init();
- return allMeasures;
- }
-
- @Override
- public boolean isReady() {
- return isReady;
- }
-
- @Override
- public String getName() {
- return name;
- }
-
- public void setName(String name) {
- this.name = name;
- }
-
- @Override
- public String toString() {
- return getCanonicalName();
- }
-
- @Override
- public String getCanonicalName() {
- return getType() + "[name=" + name + "]";
- }
-
- @Override
- public String getProjectName() {
- return projectName;
- }
-
- @Override
- public void setProjectName(String prjName) {
- projectName = prjName;
- }
-
- public KylinConfig getConfig() {
- return config;
- }
-
- public void setConfig(KylinConfig config) {
- this.config = config;
- }
-
- @Override
- public long getDateRangeStart() {
- return dateRangeStart;
- }
-
- @Override
- public long getDateRangeEnd() {
- return dateRangeEnd;
- }
-
- @Override
- public String getModelName() {
- return this.getLatestRealization().getModelName();
- }
-
- @Override
- public List<TblColRef> getAllDimensions() {
- init();
- return allDimensions;
- }
-
- public IRealization[] getRealizations() {
- init();
- return realizations;
- }
-
- public static String concatResourcePath(String hybridName) {
- return ResourceStore.HYBRID_RESOURCE_ROOT + "/" + hybridName + ".json";
- }
-
- public void setCost(int cost) {
- this.cost = cost;
- }
-
- public IRealization getLatestRealization() {
- if (realizations.length > 0) {
- return realizations[realizations.length - 1];
- }
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridManager.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridManager.java b/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridManager.java
deleted file mode 100644
index 9eb8922..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridManager.java
+++ /dev/null
@@ -1,141 +0,0 @@
-package org.apache.kylin.storage.hybrid;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.persistence.JsonSerializer;
-import org.apache.kylin.common.persistence.ResourceStore;
-import org.apache.kylin.common.persistence.Serializer;
-import org.apache.kylin.common.restclient.Broadcaster;
-import org.apache.kylin.common.restclient.CaseInsensitiveStringCache;
-import org.apache.kylin.metadata.realization.IRealization;
-import org.apache.kylin.metadata.realization.IRealizationProvider;
-import org.apache.kylin.metadata.realization.RealizationType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- */
-public class HybridManager implements IRealizationProvider {
- public static final Serializer<HybridInstance> HYBRID_SERIALIZER = new JsonSerializer<HybridInstance>(HybridInstance.class);
-
- private static final Logger logger = LoggerFactory.getLogger(HybridManager.class);
-
- // static cached instances
- private static final ConcurrentHashMap<KylinConfig, HybridManager> CACHE = new ConcurrentHashMap<KylinConfig, HybridManager>();
-
- public static HybridManager getInstance(KylinConfig config) {
- HybridManager r = CACHE.get(config);
- if (r != null) {
- return r;
- }
-
- synchronized (HybridManager.class) {
- r = CACHE.get(config);
- if (r != null) {
- return r;
- }
- try {
- r = new HybridManager(config);
- CACHE.put(config, r);
- if (CACHE.size() > 1) {
- logger.warn("More than one Hybrid Manager singleton exist");
- }
- return r;
- } catch (IOException e) {
- throw new IllegalStateException("Failed to init Hybrid Manager from " + config, e);
- }
- }
- }
-
- public static void clearCache() {
- CACHE.clear();
- }
-
- // ============================================================================
-
- private KylinConfig config;
-
- private CaseInsensitiveStringCache<HybridInstance> hybridMap = new CaseInsensitiveStringCache<HybridInstance>(Broadcaster.TYPE.HYBRID);
-
- private HybridManager(KylinConfig config) throws IOException {
- logger.info("Initializing HybridManager with config " + config);
- this.config = config;
-
- loadAllHybridInstance();
- }
-
- private void loadAllHybridInstance() throws IOException {
- ResourceStore store = getStore();
- List<String> paths = store.collectResourceRecursively(ResourceStore.HYBRID_RESOURCE_ROOT, ".json");
-
- logger.debug("Loading Hybrid from folder " + store.getReadableResourcePath(ResourceStore.HYBRID_RESOURCE_ROOT));
-
- for (String path : paths) {
- loadHybridInstance(path);
- }
-
- logger.debug("Loaded " + paths.size() + " Hybrid(s)");
- }
-
- public void reloadHybridInstanceByChild(RealizationType type, String realizationName) throws IOException {
- for (HybridInstance hybridInstance : hybridMap.values()) {
- boolean includes = false;
- for (IRealization realization : hybridInstance.getRealizations()) {
- if (realization.getType() == type && realization.getName().equalsIgnoreCase(realizationName)) {
- includes = true;
- break;
- }
- }
-
- if (includes == true)
- loadHybridInstance(HybridInstance.concatResourcePath(hybridInstance.getName()));
- }
- }
-
- private synchronized HybridInstance loadHybridInstance(String path) throws IOException {
- ResourceStore store = getStore();
-
- HybridInstance hybridInstance = null;
- try {
- hybridInstance = store.getResource(path, HybridInstance.class, HYBRID_SERIALIZER);
- hybridInstance.setConfig(config);
-
- if (hybridInstance.getRealizationEntries() == null || hybridInstance.getRealizationEntries().size() == 0) {
- throw new IllegalStateException("HybridInstance must have realization entries, " + path);
- }
-
- if (StringUtils.isBlank(hybridInstance.getName()))
- throw new IllegalStateException("HybridInstance name must not be blank, at " + path);
-
- final String name = hybridInstance.getName();
- hybridMap.putLocal(name, hybridInstance);
-
- return hybridInstance;
- } catch (Exception e) {
- logger.error("Error during load hybrid instance " + path, e);
- return null;
- }
- }
-
- @Override
- public RealizationType getRealizationType() {
- return RealizationType.HYBRID;
- }
-
- @Override
- public IRealization getRealization(String name) {
- return getHybridInstance(name);
- }
-
- public HybridInstance getHybridInstance(String name) {
- return hybridMap.get(name);
- }
-
- private ResourceStore getStore() {
- return ResourceStore.getStore(this.config);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridStorageEngine.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridStorageEngine.java b/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridStorageEngine.java
deleted file mode 100644
index 4baba93..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridStorageEngine.java
+++ /dev/null
@@ -1,44 +0,0 @@
-package org.apache.kylin.storage.hybrid;
-
-import java.util.List;
-
-import org.apache.kylin.metadata.realization.IRealization;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.tuple.CompoundTupleIterator;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.storage.IStorageQuery;
-import org.apache.kylin.storage.StorageContext;
-import org.apache.kylin.storage.StorageFactory;
-import org.apache.kylin.storage.tuple.TupleInfo;
-
-import com.google.common.collect.Lists;
-
-/**
- */
-public class HybridStorageEngine implements IStorageQuery {
-
- private IRealization[] realizations;
- private IStorageQuery[] storageEngines;
-
- public HybridStorageEngine(HybridInstance hybridInstance) {
- this.realizations = hybridInstance.getRealizations();
- storageEngines = new IStorageQuery[realizations.length];
- for (int i = 0; i < realizations.length; i++) {
- storageEngines[i] = StorageFactory.createQuery(realizations[i]);
- }
- }
-
- @Override
- public ITupleIterator search(final StorageContext context, final SQLDigest sqlDigest, final TupleInfo returnTupleInfo) {
- List<ITupleIterator> tupleIterators = Lists.newArrayList();
- for (int i = 0; i < realizations.length; i++) {
- if (realizations[i].isReady() && realizations[i].isCapable(sqlDigest)) {
- ITupleIterator dataIterator = storageEngines[i].search(context, sqlDigest, returnTupleInfo);
- tupleIterators.add(dataIterator);
- }
- }
- // combine tuple iterator
- return new CompoundTupleIterator(tupleIterators);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/main/java/org/apache/kylin/storage/util/SizeOfUtil.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/util/SizeOfUtil.java b/storage/src/main/java/org/apache/kylin/storage/util/SizeOfUtil.java
deleted file mode 100644
index ccbe845..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/util/SizeOfUtil.java
+++ /dev/null
@@ -1,21 +0,0 @@
-package org.apache.kylin.storage.util;
-
-import net.sf.ehcache.pool.sizeof.ReflectionSizeOf;
-
-/**
- */
-public final class SizeOfUtil {
-
- private SizeOfUtil() {
- }
-
- private static final ReflectionSizeOf DEFAULT_SIZE_OF = new ReflectionSizeOf();
-
- public static final long deepSizeOf(Object obj) {
- return DEFAULT_SIZE_OF.deepSizeOf(Integer.MAX_VALUE, true, obj).getCalculated();
- }
-
- public static final long sizeOf(Object obj) {
- return DEFAULT_SIZE_OF.sizeOf(obj);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/test/java/org/apache/kylin/storage/MiniClusterTest.java
----------------------------------------------------------------------
diff --git a/storage/src/test/java/org/apache/kylin/storage/MiniClusterTest.java b/storage/src/test/java/org/apache/kylin/storage/MiniClusterTest.java
deleted file mode 100644
index 9ee6e94..0000000
--- a/storage/src/test/java/org/apache/kylin/storage/MiniClusterTest.java
+++ /dev/null
@@ -1,63 +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;
-
-import java.io.File;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseTestingUtility;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.MiniHBaseCluster;
-import org.apache.kylin.common.util.AbstractKylinTestCase;
-
-public class MiniClusterTest {
-
- private static HBaseTestingUtility testUtil = new HBaseTestingUtility();
-
- public static void main(String[] args) throws Exception {
-
- File miniclusterFolder = new File(AbstractKylinTestCase.MINICLUSTER_TEST_DATA);
- System.out.println("----" + miniclusterFolder.getAbsolutePath());
-
- //save the dfs data to minicluster folder
- System.setProperty("test.build.data", miniclusterFolder.getAbsolutePath());
-
- MiniHBaseCluster hbCluster = testUtil.startMiniCluster(1);
- testUtil.startMiniMapReduceCluster();
- System.out.println("Minicluster started.");
-
- Configuration conf = hbCluster.getConf();
- String host = conf.get(HConstants.ZOOKEEPER_QUORUM);
- String port = conf.get(HConstants.ZOOKEEPER_CLIENT_PORT);
- String parent = conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT);
-
- // reduce rpc retry
- conf.set(HConstants.HBASE_CLIENT_PAUSE, "3000");
- conf.set(HConstants.HBASE_CLIENT_RETRIES_NUMBER, "5");
- conf.set(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, "60000");
-
- String connectionUrl = "hbase:" + host + ":" + port + ":" + parent;
-
- System.out.println("hbase connection url:" + connectionUrl);
-
- testUtil.getDFSCluster().getFileSystem();
- testUtil.shutdownMiniMapReduceCluster();
- testUtil.shutdownMiniCluster();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java
----------------------------------------------------------------------
diff --git a/storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java b/storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java
deleted file mode 100644
index 0279c8c..0000000
--- a/storage/src/test/java/org/apache/kylin/storage/cache/EhcacheTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-package org.apache.kylin.storage.cache;
-
-import java.util.List;
-
-import net.sf.ehcache.Cache;
-import net.sf.ehcache.CacheManager;
-import net.sf.ehcache.Element;
-import net.sf.ehcache.config.CacheConfiguration;
-import net.sf.ehcache.config.Configuration;
-import net.sf.ehcache.config.PersistenceConfiguration;
-import net.sf.ehcache.store.MemoryStoreEvictionPolicy;
-
-import org.junit.Test;
-
-import com.google.common.collect.Lists;
-
-/**
- */
-public class EhcacheTest {
- @Test
- public void basicTest() throws InterruptedException {
- System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
-
- Configuration conf = new Configuration();
- conf.setMaxBytesLocalHeap("100M");
- CacheManager cacheManager = CacheManager.create(conf);
-
- //Create a Cache specifying its configuration.
- Cache testCache = //Create a Cache specifying its configuration.
- new Cache(new CacheConfiguration("test", 0).//
- memoryStoreEvictionPolicy(MemoryStoreEvictionPolicy.LRU).//
- eternal(false).//
- timeToIdleSeconds(86400).//
- diskExpiryThreadIntervalSeconds(0).//
- //maxBytesLocalHeap(1000, MemoryUnit.MEGABYTES).//
- persistence(new PersistenceConfiguration().strategy(PersistenceConfiguration.Strategy.NONE)));
-
- cacheManager.addCache(testCache);
-
- System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
- byte[] blob = new byte[(1024 * 40 * 1024)];//400M
-
- List<String> manyObjects = Lists.newArrayList();
- for (int i = 0; i < 10000; i++) {
- manyObjects.add(new String("" + i));
- }
- testCache.put(new Element("0", manyObjects));
-
- testCache.put(new Element("1", blob));
- System.out.println(testCache.get("1") == null);
- System.out.println(testCache.getSize());
- System.out.println(testCache.getStatistics().getLocalHeapSizeInBytes());
- System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
- testCache.put(new Element("2", blob));
- System.out.println(testCache.get("1") == null);
- System.out.println(testCache.getSize());
- System.out.println(testCache.getStatistics().getLocalHeapSizeInBytes());
- System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
- testCache.put(new Element("3", blob));
- System.out.println(testCache.get("1") == null);
- System.out.println(testCache.get("2") == null);
- System.out.println(testCache.get("3") == null);
- System.out.println(testCache.getSize());
- System.out.println(testCache.getStatistics().getLocalHeapSizeInBytes());
- System.out.println("runtime used memory: " + (Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory()) / 1024 / 1024 + "M");
-
- cacheManager.shutdown();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/test/java/org/apache/kylin/storage/filter/BitMapFilterEvaluatorTest.java
----------------------------------------------------------------------
diff --git a/storage/src/test/java/org/apache/kylin/storage/filter/BitMapFilterEvaluatorTest.java b/storage/src/test/java/org/apache/kylin/storage/filter/BitMapFilterEvaluatorTest.java
deleted file mode 100644
index 3aca8f8..0000000
--- a/storage/src/test/java/org/apache/kylin/storage/filter/BitMapFilterEvaluatorTest.java
+++ /dev/null
@@ -1,231 +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 static org.junit.Assert.assertEquals;
-
-import java.util.ArrayList;
-
-import org.apache.kylin.dict.Dictionary;
-import org.apache.kylin.metadata.filter.ColumnTupleFilter;
-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.filter.TupleFilter.FilterOperatorEnum;
-import org.apache.kylin.metadata.model.ColumnDesc;
-import org.apache.kylin.metadata.model.TableDesc;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.storage.filter.BitMapFilterEvaluator.BitMapProvider;
-import org.junit.Test;
-
-import com.google.common.collect.Lists;
-import it.uniroma3.mat.extendedset.intset.ConciseSet;
-
-public class BitMapFilterEvaluatorTest {
-
- static TblColRef colA;
- static TblColRef colB;
-
- static {
- TableDesc table = TableDesc.mockup("DEFAULT.TABLE");
-
- ColumnDesc col = ColumnDesc.mockup(table, 1, "colA", "string");
- colA = new TblColRef(col);
-
- col = ColumnDesc.mockup(table, 1, "colB", "string");
- colB = new TblColRef(col);
- }
-
- static class MockBitMapProivder implements BitMapProvider {
-
- private static final int MAX_ID = 8;
- private static final int REC_COUNT = 10;
-
- @Override
- public ConciseSet getBitMap(TblColRef col, Integer startId, Integer endId) {
- if (!col.equals(colA))
- return null;
-
- // i-th record has value ID i, and last record has value null
- if (startId == null && endId == null) {
- //entry for getting null value
- ConciseSet s = new ConciseSet();
- s.add(getRecordCount() - 1);
- return s;
- }
-
- int start = 0;
- int end = MAX_ID;
- if (startId != null) {
- start = startId;
- }
- if (endId != null) {
- end = endId;
- }
-
- ConciseSet ret = new ConciseSet();
- for (int i = start; i <= end; ++i) {
- ConciseSet temp = getBitMap(col, i);
- ret.addAll(temp);
- }
- return ret;
- }
-
- public ConciseSet getBitMap(TblColRef col, int valueId) {
- if (!col.equals(colA))
- return null;
-
- // i-th record has value ID i, and last record has value null
- ConciseSet bitMap = new ConciseSet();
- if (valueId < 0 || valueId > getMaxValueId(col)) // null
- bitMap.add(getRecordCount() - 1);
- else
- bitMap.add(valueId);
-
- return bitMap;
- }
-
- @Override
- public int getRecordCount() {
- return REC_COUNT;
- }
-
- @Override
- public int getMaxValueId(TblColRef col) {
- return MAX_ID;
- }
- }
-
- BitMapFilterEvaluator eval = new BitMapFilterEvaluator(new MockBitMapProivder());
- ArrayList<CompareTupleFilter> basicFilters = Lists.newArrayList();
- ArrayList<ConciseSet> basicResults = Lists.newArrayList();
-
- public BitMapFilterEvaluatorTest() {
- basicFilters.add(compare(colA, FilterOperatorEnum.ISNULL));
- basicResults.add(set(9));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.ISNOTNULL));
- basicResults.add(set(0, 1, 2, 3, 4, 5, 6, 7, 8));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.EQ, 0));
- basicResults.add(set(0));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.NEQ, 0));
- basicResults.add(set(1, 2, 3, 4, 5, 6, 7, 8));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.IN, 0, 5));
- basicResults.add(set(0, 5));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.NOTIN, 0, 5));
- basicResults.add(set(1, 2, 3, 4, 6, 7, 8));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.LT, 3));
- basicResults.add(set(0, 1, 2));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.LTE, 3));
- basicResults.add(set(0, 1, 2, 3));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.GT, 3));
- basicResults.add(set(4, 5, 6, 7, 8));
-
- basicFilters.add(compare(colA, FilterOperatorEnum.GTE, 3));
- basicResults.add(set(3, 4, 5, 6, 7, 8));
- }
-
- @Test
- public void testBasics() {
- for (int i = 0; i < basicFilters.size(); i++) {
- assertEquals(basicResults.get(i), eval.evaluate(basicFilters.get(i)));
- }
- }
-
- @Test
- public void testLogicalAnd() {
- for (int i = 0; i < basicFilters.size(); i++) {
- for (int j = 0; j < basicFilters.size(); j++) {
- LogicalTupleFilter f = logical(FilterOperatorEnum.AND, basicFilters.get(i), basicFilters.get(j));
- ConciseSet r = basicResults.get(i).clone();
- r.retainAll(basicResults.get(j));
- assertEquals(r, eval.evaluate(f));
- }
- }
- }
-
- @Test
- public void testLogicalOr() {
- for (int i = 0; i < basicFilters.size(); i++) {
- for (int j = 0; j < basicFilters.size(); j++) {
- LogicalTupleFilter f = logical(FilterOperatorEnum.OR, basicFilters.get(i), basicFilters.get(j));
- ConciseSet r = basicResults.get(i).clone();
- r.addAll(basicResults.get(j));
- assertEquals(r, eval.evaluate(f));
- }
- }
- }
-
- @Test
- public void testNotEvaluable() {
- CompareTupleFilter notEvaluable = compare(colB, FilterOperatorEnum.EQ, 0);
- assertEquals(null, eval.evaluate(notEvaluable));
-
- LogicalTupleFilter or = logical(FilterOperatorEnum.OR, basicFilters.get(1), notEvaluable);
- assertEquals(null, eval.evaluate(or));
-
- LogicalTupleFilter and = logical(FilterOperatorEnum.AND, basicFilters.get(1), notEvaluable);
- assertEquals(basicResults.get(1), eval.evaluate(and));
- }
-
- public static CompareTupleFilter compare(TblColRef col, TupleFilter.FilterOperatorEnum op, int... ids) {
- CompareTupleFilter filter = new CompareTupleFilter(op);
- filter.addChild(columnFilter(col));
- for (int i : ids) {
- filter.addChild(constFilter(i));
- }
- return filter;
- }
-
- public static LogicalTupleFilter logical(TupleFilter.FilterOperatorEnum op, TupleFilter... filters) {
- LogicalTupleFilter filter = new LogicalTupleFilter(op);
- for (TupleFilter f : filters)
- filter.addChild(f);
- return filter;
- }
-
- public static ColumnTupleFilter columnFilter(TblColRef col) {
- return new ColumnTupleFilter(col);
- }
-
- public static ConstantTupleFilter constFilter(int id) {
- return new ConstantTupleFilter(idToStr(id));
- }
-
- public static ConciseSet set(int... ints) {
- ConciseSet set = new ConciseSet();
- for (int i : ints)
- set.add(i);
- return set;
- }
-
- public static String idToStr(int id) {
- byte[] bytes = new byte[] { (byte) id };
- return Dictionary.dictIdToString(bytes, 0, bytes.length);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/11611cc2/storage/src/test/java/org/apache/kylin/storage/filter/FilterBaseTest.java
----------------------------------------------------------------------
diff --git a/storage/src/test/java/org/apache/kylin/storage/filter/FilterBaseTest.java b/storage/src/test/java/org/apache/kylin/storage/filter/FilterBaseTest.java
deleted file mode 100644
index 12c64ce..0000000
--- a/storage/src/test/java/org/apache/kylin/storage/filter/FilterBaseTest.java
+++ /dev/null
@@ -1,219 +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.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Random;
-
-import org.apache.kylin.metadata.filter.CaseTupleFilter;
-import org.apache.kylin.metadata.filter.ColumnTupleFilter;
-import org.apache.kylin.metadata.filter.CompareTupleFilter;
-import org.apache.kylin.metadata.filter.ConstantTupleFilter;
-import org.apache.kylin.metadata.filter.DynamicTupleFilter;
-import org.apache.kylin.metadata.filter.IFilterCodeSystem;
-import org.apache.kylin.metadata.filter.LogicalTupleFilter;
-import org.apache.kylin.metadata.filter.StringCodeSystem;
-import org.apache.kylin.metadata.filter.TupleFilter;
-import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
-import org.apache.kylin.metadata.model.ColumnDesc;
-import org.apache.kylin.metadata.model.TableDesc;
-import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.storage.tuple.Tuple;
-import org.apache.kylin.storage.tuple.TupleInfo;
-
-/**
- * @author xjiang
- *
- */
-public class FilterBaseTest {
-
- @SuppressWarnings("rawtypes")
- static final IFilterCodeSystem CS = StringCodeSystem.INSTANCE;
-
- protected List<TblColRef> buildGroups() {
- List<TblColRef> groups = new ArrayList<TblColRef>();
-
- TableDesc t1 = TableDesc.mockup("DEFAULT.TEST_KYLIN_FACT");
- ColumnDesc c1 = ColumnDesc.mockup(t1, 2, "CAL_DT", "string");
- TblColRef cf1 = new TblColRef(c1);
- groups.add(cf1);
-
- TableDesc t2 = TableDesc.mockup("DEFAULT.TEST_CATEGORY_GROUPINGS");
- ColumnDesc c2 = ColumnDesc.mockup(t2, 14, "META_CATEG_NAME", "string");
- TblColRef cf2 = new TblColRef(c2);
- groups.add(cf2);
-
- return groups;
- }
-
- protected CompareTupleFilter buildCompareFilter(List<TblColRef> groups, int index) {
- TblColRef column = groups.get(index);
- CompareTupleFilter compareFilter = new CompareTupleFilter(FilterOperatorEnum.EQ);
- ColumnTupleFilter columnFilter = new ColumnTupleFilter(column);
- compareFilter.addChild(columnFilter);
- ConstantTupleFilter constantFilter = null;
- if (index == 0) {
- constantFilter = new ConstantTupleFilter("2013-03-10");
- } else if (index == 1) {
- constantFilter = new ConstantTupleFilter("ClothinShoes & Accessories");
- }
- compareFilter.addChild(constantFilter);
- return compareFilter;
- }
-
- protected TupleFilter buildAndFilter(List<TblColRef> columns) {
- CompareTupleFilter compareFilter1 = buildCompareFilter(columns, 0);
- CompareTupleFilter compareFilter2 = buildCompareFilter(columns, 1);
- LogicalTupleFilter andFilter = new LogicalTupleFilter(FilterOperatorEnum.AND);
- andFilter.addChild(compareFilter1);
- andFilter.addChild(compareFilter2);
- return andFilter;
- }
-
- protected TupleFilter buildOrFilter(List<TblColRef> columns) {
- CompareTupleFilter compareFilter1 = buildCompareFilter(columns, 0);
- CompareTupleFilter compareFilter2 = buildCompareFilter(columns, 1);
- LogicalTupleFilter logicFilter = new LogicalTupleFilter(FilterOperatorEnum.OR);
- logicFilter.addChild(compareFilter1);
- logicFilter.addChild(compareFilter2);
- return logicFilter;
- }
-
- protected CaseTupleFilter buildCaseFilter(List<TblColRef> groups) {
- CaseTupleFilter caseFilter = new CaseTupleFilter();
-
- TupleFilter when0 = buildAndFilter(groups);
- caseFilter.addChild(when0);
- TupleFilter then0 = new ConstantTupleFilter("0");
- caseFilter.addChild(then0);
-
- TupleFilter when1 = buildCompareFilter(groups, 0);
- caseFilter.addChild(when1);
- TupleFilter then1 = new ConstantTupleFilter("1");
- caseFilter.addChild(then1);
-
- TupleFilter when2 = buildCompareFilter(groups, 1);
- caseFilter.addChild(when2);
- TupleFilter then2 = new ConstantTupleFilter("2");
- caseFilter.addChild(then2);
-
- TupleFilter when3 = buildOrFilter(groups);
- caseFilter.addChild(when3);
- TupleFilter then3 = new ConstantTupleFilter("3");
- caseFilter.addChild(then3);
-
- TupleFilter else4 = new ConstantTupleFilter("4");
- caseFilter.addChild(else4);
-
- return caseFilter;
- }
-
- protected CompareTupleFilter buildCompareCaseFilter(List<TblColRef> groups, String constValue) {
- CompareTupleFilter compareFilter = new CompareTupleFilter(FilterOperatorEnum.EQ);
- CaseTupleFilter caseFilter = buildCaseFilter(groups);
- compareFilter.addChild(caseFilter);
- ConstantTupleFilter constantFilter = new ConstantTupleFilter(constValue);
- compareFilter.addChild(constantFilter);
- return compareFilter;
- }
-
- protected CompareTupleFilter buildCompareDynamicFilter(List<TblColRef> groups) {
- CompareTupleFilter compareFilter = new CompareTupleFilter(FilterOperatorEnum.EQ);
- compareFilter.addChild(new ColumnTupleFilter(groups.get(0)));
- compareFilter.addChild(new DynamicTupleFilter("?0"));
- compareFilter.bindVariable("?0", "abc");
- return compareFilter;
- }
-
- protected void compareFilter(TupleFilter f1, TupleFilter f2) {
- if (f1 == null && f2 == null) {
- return;
- }
-
- if (f1 == null || f2 == null) {
- throw new IllegalStateException("f1=" + f1 + ", f2=" + f2);
- }
-
- String str1 = f1.toString();
- System.out.println("f1=" + str1);
- String str2 = f2.toString();
- System.out.println("f2=" + str2);
- if (!str1.equals(str2)) {
- throw new IllegalStateException("f1=" + str1 + ", f2=" + str2);
- }
-
- int s1 = f1.getChildren().size();
- int s2 = f2.getChildren().size();
- if (s1 != s2) {
- throw new IllegalStateException("f1=" + str1 + ", f2=" + str2 + " has different children: " + s1 + " vs. " + s2);
- }
-
- for (int i = 0; i < s1; i++) {
- compareFilter(f1.getChildren().get(i), f2.getChildren().get(i));
- }
- }
-
- private static String[][] SAMPLE_DATA = new String[][] { { "2013-03-10", "2012-01-12", "2014-03-10" }, { "ClothinShoes & Accessories", "ABIN", "FP-GTC", "FP-NON-GTC" } };
-
- protected Collection<Tuple> generateTuple(int number, List<TblColRef> columns, int[] matches) {
-
- Collection<Tuple> tuples = new ArrayList<Tuple>(number);
- TupleInfo info = new TupleInfo();
- for (int i = 0; i < columns.size(); i++) {
- TblColRef column = columns.get(i);
- info.setField(column.getName(), column, i);
- }
-
- int allMatches = 0;
- Random rand = new Random();
- for (int i = 0; i < number; i++) {
- Tuple t = new Tuple(info);
- boolean isFullMatch = true;
- for (int k = 0; k < columns.size(); k++) {
- TblColRef column = columns.get(k);
- int index = Math.abs(rand.nextInt()) % SAMPLE_DATA[k].length;
- t.setDimensionValue(column.getName(), SAMPLE_DATA[k][index]);
- if (index == 0) {
- matches[k]++;
- } else {
- isFullMatch = false;
- }
- }
- if (isFullMatch) {
- allMatches++;
- }
- tuples.add(t);
- }
- matches[2] = allMatches;
- return tuples;
- }
-
- protected int evaluateTuples(Collection<Tuple> tuples, TupleFilter filter) {
- int match = 0;
- for (Tuple t : tuples) {
- if (filter.evaluate(t, CS)) {
- match++;
- }
- }
- return match;
- }
-
-}