You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/04/26 06:27:17 UTC
[35/50] incubator-kylin git commit: KYLIN-671 shrink cachable tsRange
with respect to each region's build status
KYLIN-671 shrink cachable tsRange with respect to each region's build status
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/fba32d88
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/fba32d88
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/fba32d88
Branch: refs/heads/streaming-localdict
Commit: fba32d881a64cc1be0cc3a5b07c05eb2b7d9fd1d
Parents: 724660a
Author: honma <ho...@ebay.com>
Authored: Tue Apr 21 17:25:28 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Wed Apr 22 16:46:02 2015 +0800
----------------------------------------------------------------------
.../metadata/realization/SQLDigestUtil.java | 1 +
.../metadata/tuple/CompoundTupleIterator.java | 24 ++++
.../kylin/metadata/tuple/ITupleIterator.java | 14 +++
.../metadata/tuple/SimpleTupleIterator.java | 7 ++
.../kylin/metadata/tuple/TeeTupleIterator.java | 13 ++-
.../test/resources/query/sql_cache/query09.sql | 11 ++
.../cache/CacheFledgedStorageEngine.java | 38 +++++--
.../kylin/storage/cache/StreamSQLResult.java | 28 +++--
.../cube/SerializedCubeTupleIterator.java | 6 +
.../storage/hbase/CubeSegmentTupleIterator.java | 6 +
.../hbase/SerializedHBaseTupleIterator.java | 6 +
.../endpoint/EndpointTupleIterator.java | 55 +++++++---
.../hbase/coprocessor/endpoint/IIEndpoint.java | 41 +++----
.../endpoint/generated/IIProtos.java | 110 ++-----------------
.../coprocessor/endpoint/protobuf/II.proto | 1 -
.../org/apache/kylin/storage/tuple/Tuple.java | 14 ++-
16 files changed, 213 insertions(+), 162 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigestUtil.java
----------------------------------------------------------------------
diff --git a/metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigestUtil.java b/metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigestUtil.java
index d15b3e3..3e1426f 100644
--- a/metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigestUtil.java
+++ b/metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigestUtil.java
@@ -44,6 +44,7 @@ public class SQLDigestUtil {
return ret;
}
+ //ts column type differentiate
private static String formatTimeStr(DataType type, long ts) {
String ret;
if (type == DataType.getInstance("date")) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/metadata/src/main/java/org/apache/kylin/metadata/tuple/CompoundTupleIterator.java
----------------------------------------------------------------------
diff --git a/metadata/src/main/java/org/apache/kylin/metadata/tuple/CompoundTupleIterator.java b/metadata/src/main/java/org/apache/kylin/metadata/tuple/CompoundTupleIterator.java
index 3b35ac2..7803d0d 100644
--- a/metadata/src/main/java/org/apache/kylin/metadata/tuple/CompoundTupleIterator.java
+++ b/metadata/src/main/java/org/apache/kylin/metadata/tuple/CompoundTupleIterator.java
@@ -3,16 +3,22 @@ package org.apache.kylin.metadata.tuple;
import java.util.Iterator;
import java.util.List;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Iterators;
+import com.google.common.collect.Range;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Created by Hongbin Ma(Binmahone) on 4/14/15.
*/
public class CompoundTupleIterator implements ITupleIterator {
+ private static final Logger logger = LoggerFactory.getLogger(CompoundTupleIterator.class);
private List<ITupleIterator> backends;
private Iterator<ITuple> compoundIterator;
public CompoundTupleIterator(List<ITupleIterator> backends) {
+ Preconditions.checkArgument(backends != null && backends.size() != 0, "backends not exists");
this.backends = backends;
this.compoundIterator = Iterators.concat(backends.iterator());
}
@@ -25,6 +31,24 @@ public class CompoundTupleIterator implements ITupleIterator {
}
@Override
+ public Range<Long> getCacheExcludedPeriod() {
+ Range<Long> timeSpan = null;
+ for (ITupleIterator itt : this.backends) {
+ Range<Long> excluded = itt.getCacheExcludedPeriod();
+ if (excluded != null && !excluded.isEmpty()) {
+ if (timeSpan == null) {
+ //first one
+ timeSpan = excluded;
+ } else {
+ logger.warn("There are two excluded period, use a span to replace them: " + timeSpan + " and " + excluded );
+ timeSpan = timeSpan.span(excluded);
+ }
+ }
+ }
+ return timeSpan;
+ }
+
+ @Override
public boolean hasNext() {
return this.compoundIterator.hasNext();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/metadata/src/main/java/org/apache/kylin/metadata/tuple/ITupleIterator.java
----------------------------------------------------------------------
diff --git a/metadata/src/main/java/org/apache/kylin/metadata/tuple/ITupleIterator.java b/metadata/src/main/java/org/apache/kylin/metadata/tuple/ITupleIterator.java
index 73cd75c..9e92eb9 100644
--- a/metadata/src/main/java/org/apache/kylin/metadata/tuple/ITupleIterator.java
+++ b/metadata/src/main/java/org/apache/kylin/metadata/tuple/ITupleIterator.java
@@ -20,6 +20,8 @@ package org.apache.kylin.metadata.tuple;
import java.util.Iterator;
+import com.google.common.collect.Range;
+
/**
* @author xjiang
*
@@ -45,11 +47,23 @@ public interface ITupleIterator extends Iterator<ITuple> {
public void close() {
}
+ @Override
+ public Range<Long> getCacheExcludedPeriod() {
+ return null;
+ }
};
void close();
/**
+ * 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
+ */
+ Range<Long> getCacheExcludedPeriod();
+
+ /**
* if hasNext() returns false because there's no more data, return true
* if hasNext() returns false because limits or threshold, return false
* if hasNext() returns true, throw IllegalStateException
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/metadata/src/main/java/org/apache/kylin/metadata/tuple/SimpleTupleIterator.java
----------------------------------------------------------------------
diff --git a/metadata/src/main/java/org/apache/kylin/metadata/tuple/SimpleTupleIterator.java b/metadata/src/main/java/org/apache/kylin/metadata/tuple/SimpleTupleIterator.java
index b3f0caa..bbad038 100644
--- a/metadata/src/main/java/org/apache/kylin/metadata/tuple/SimpleTupleIterator.java
+++ b/metadata/src/main/java/org/apache/kylin/metadata/tuple/SimpleTupleIterator.java
@@ -2,6 +2,8 @@ package org.apache.kylin.metadata.tuple;
import java.util.Iterator;
+import com.google.common.collect.Range;
+
/**
* Created by Hongbin Ma(Binmahone) on 4/14/15.
*
@@ -32,4 +34,9 @@ public class SimpleTupleIterator implements ITupleIterator {
@Override
public void close() {
}
+
+ @Override
+ public Range<Long> getCacheExcludedPeriod() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java
----------------------------------------------------------------------
diff --git a/metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java b/metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java
index 0314762..85465b1 100644
--- a/metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java
+++ b/metadata/src/main/java/org/apache/kylin/metadata/tuple/TeeTupleIterator.java
@@ -5,6 +5,7 @@ import java.util.List;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
+import com.google.common.collect.Range;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -22,12 +23,15 @@ public class TeeTupleIterator implements ITupleIterator {
private ITupleIterator underlying;
private List<ITuple> duplicatedData;
- public TeeTupleIterator(ITupleIterator underlying, Function<List<ITuple>, Void> actionOnSeeingWholeData) {
- this.actionOnSeeingWholeData = actionOnSeeingWholeData;
+ public TeeTupleIterator(ITupleIterator underlying) {
this.underlying = underlying;
this.duplicatedData = Lists.newArrayList();
}
+ public void setActionOnSeeingWholeData(Function<List<ITuple>, Void> actionOnSeeingWholeData) {
+ this.actionOnSeeingWholeData = actionOnSeeingWholeData;
+ }
+
@Override
public void close() {
this.underlying.close();
@@ -36,6 +40,11 @@ public class TeeTupleIterator implements ITupleIterator {
}
@Override
+ public Range<Long> getCacheExcludedPeriod() {
+ return this.underlying.getCacheExcludedPeriod();
+ }
+
+ @Override
public boolean hasNext() {
return this.underlying.hasNext();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/query/src/test/resources/query/sql_cache/query09.sql
----------------------------------------------------------------------
diff --git a/query/src/test/resources/query/sql_cache/query09.sql b/query/src/test/resources/query/sql_cache/query09.sql
new file mode 100644
index 0000000..b9e2821
--- /dev/null
+++ b/query/src/test/resources/query/sql_cache/query09.sql
@@ -0,0 +1,11 @@
+select test_kylin_fact.lstg_format_name, test_kylin_fact.cal_dt,sum(test_kylin_fact.price) as GMV
+ , count(*) as TRANS_CNT
+ from test_kylin_fact
+inner JOIN edw.test_cal_dt as test_cal_dt
+ ON test_kylin_fact.cal_dt = test_cal_dt.cal_dt
+ inner JOIN test_category_groupings
+ ON test_kylin_fact.leaf_categ_id = test_category_groupings.leaf_categ_id AND test_kylin_fact.lstg_site_id = test_category_groupings.site_id
+ inner JOIN edw.test_sites as test_sites
+ ON test_kylin_fact.lstg_site_id = test_sites.site_id
+ where test_kylin_fact.cal_dt < DATE '2018-10-10' AND test_kylin_fact.cal_dt > DATE '2009-09-10'
+ group by test_kylin_fact.lstg_format_name, test_kylin_fact.cal_dt
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStorageEngine.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStorageEngine.java b/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStorageEngine.java
index 7a2aa94..4f8ad2f 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStorageEngine.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedStorageEngine.java
@@ -4,6 +4,7 @@ import java.util.List;
import javax.annotation.Nullable;
+import com.google.common.collect.Ranges;
import net.sf.ehcache.Cache;
import net.sf.ehcache.CacheManager;
import net.sf.ehcache.Element;
@@ -82,7 +83,8 @@ public class CacheFledgedStorageEngine implements IStorageEngine {
boolean needUpdateCache = true;
final StreamSQLDigest streamSQLDigest = new StreamSQLDigest(sqlDigest, partitionColRef);
StreamSQLResult cachedResult = null;
- Element element = cacheManager.getCache(STORAGE_LAYER_TUPLE_CACHE).get(streamSQLDigest);
+ Cache cache = cacheManager.getCache(STORAGE_LAYER_TUPLE_CACHE);
+ Element element = cache.get(streamSQLDigest);
if (element != null) {
cachedResult = (StreamSQLResult) element.getObjectValue();
}
@@ -93,19 +95,17 @@ public class CacheFledgedStorageEngine implements IStorageEngine {
return ITupleIterator.EMPTY_TUPLE_ITERATOR;
}
- final Range<Long> tsRange = ts;
-
ITupleIterator ret = null;
if (cachedResult != null) {
logger.debug("existing cache : " + cachedResult);
- Range<Long> reusePeriod = cachedResult.getReusableResults(tsRange);
+ Range<Long> reusePeriod = cachedResult.getReusableResults(ts);
- logger.info("ts Range in query: " + RangeUtil.formatTsRange(tsRange));
+ logger.info("ts Range in query: " + RangeUtil.formatTsRange(ts));
logger.info("potential reusable range : " + RangeUtil.formatTsRange(reusePeriod));
if (reusePeriod != null) {
- List<Range<Long>> remainings = RangeUtil.remove(tsRange, reusePeriod);
+ List<Range<Long>> remainings = RangeUtil.remove(ts, reusePeriod);
if (remainings.size() == 1) {
SimpleTupleIterator reusedTuples = new SimpleTupleIterator(cachedResult.reuse(reusePeriod));
@@ -124,6 +124,8 @@ public class CacheFledgedStorageEngine implements IStorageEngine {
}
//if remaining size > 1, we skip using cache , i.e, ret will == null
}
+ } else {
+ logger.info("no cache entry for this query");
}
if (ret == null) {
@@ -135,15 +137,33 @@ public class CacheFledgedStorageEngine implements IStorageEngine {
}
if (needUpdateCache) {
+ //the tsRange in cache should reflect data aliveness
+ final Range<Long> finalTs = ts;
+
//use another nested ITupleIterator to deal with cache
- ITupleIterator tee = new TeeTupleIterator(ret, new Function<List<ITuple>, Void>() {
+ final TeeTupleIterator tee = new TeeTupleIterator(ret);
+ tee.setActionOnSeeingWholeData(new Function<List<ITuple>, Void>() {
@Nullable
@Override
public Void apply(List<ITuple> input) {
- //TODO: tsRange needs updated
+ Range<Long> tsRange = finalTs;
+ Range<Long> cacheExclude = tee.getCacheExcludedPeriod();
+ if (cacheExclude != null) {
+ List<Range<Long>> cachablePeriods = RangeUtil.remove(tsRange, cacheExclude);
+ if (cachablePeriods.size() == 1) {
+ if (!tsRange.equals(cachablePeriods.get(0))) {
+ logger.info("With respect to each shard's build status, the cachable tsRange shrinks from " + RangeUtil.formatTsRange(tsRange) + " to " + RangeUtil.formatTsRange(cachablePeriods.get(0)));
+ }
+ tsRange = cachablePeriods.get(0);
+ } else {
+ //give up updating the cache, in avoid to make cache complicated
+ return null;
+ }
+ }
+
StreamSQLResult newCacheEntry = new StreamSQLResult(input, tsRange, partitionColRef);
cacheManager.getCache(STORAGE_LAYER_TUPLE_CACHE).put(new Element(streamSQLDigest, newCacheEntry));
- logger.debug("current cache: " + newCacheEntry);
+ logger.debug("cache after the query: " + newCacheEntry);
return null;
}
});
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java b/storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java
index 83486f9..52405d8 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cache/StreamSQLResult.java
@@ -6,6 +6,7 @@ import java.util.NavigableMap;
import javax.annotation.Nullable;
+import com.google.common.base.Preconditions;
import org.apache.kylin.common.util.RangeUtil;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.tuple.ITuple;
@@ -26,21 +27,21 @@ public class StreamSQLResult {
private NavigableMap<Long, List<ITuple>> sortedRows;
public StreamSQLResult(List<ITuple> rows, Range<Long> timeCovered, TblColRef partitionCol) {
+
+ Preconditions.checkArgument(timeCovered.hasUpperBound(),"StreamSQLResult requires timeCovered having a upperBound");
+
sortedRows = Maps.newTreeMap();
for (ITuple row : rows) {
- //TODO: differeniate between partitionCol types
- long t;
- if (partitionCol.getDatatype().equals("date")) {
- t = Tuple.epicDaysToMillis(Integer.valueOf(row.getValue(partitionCol).toString()));
- } else {
- t = Long.valueOf(row.getValue(partitionCol).toString());
- }
+ long t = Tuple.getTs(row,partitionCol);
- if (!this.sortedRows.containsKey(t)) {
- this.sortedRows.put(t, Lists.newArrayList(row));
- } else {
- this.sortedRows.get(t).add(row);
+ //will only cache rows that are within the time range
+ if (timeCovered.contains(t)) {
+ if (!this.sortedRows.containsKey(t)) {
+ this.sortedRows.put(t, Lists.newArrayList(row));
+ } else {
+ this.sortedRows.get(t).add(row);
+ }
}
}
this.timeCovered = timeCovered;
@@ -75,4 +76,9 @@ public class StreamSQLResult {
public String toString() {
return sortedRows.size() + " tuples cached for period " + RangeUtil.formatTsRange(timeCovered);
}
+
+ public long getEndTime()
+ {
+ return this.timeCovered.upperEndpoint();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java
index e0605b4..e69a1c7 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/SerializedCubeTupleIterator.java
@@ -2,6 +2,7 @@ package org.apache.kylin.storage.cube;
import java.util.List;
+import com.google.common.collect.Range;
import org.apache.kylin.metadata.tuple.ITuple;
import org.apache.kylin.metadata.tuple.ITupleIterator;
@@ -34,4 +35,9 @@ public class SerializedCubeTupleIterator implements ITupleIterator {
}
+ @Override
+ public Range<Long> getCacheExcludedPeriod() {
+ return null;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java
index 937daad..a15f372 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/CubeSegmentTupleIterator.java
@@ -23,6 +23,7 @@ import java.text.MessageFormat;
import java.util.*;
import java.util.Map.Entry;
+import com.google.common.collect.Range;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
import org.apache.hadoop.hbase.filter.Filter;
@@ -110,6 +111,11 @@ public class CubeSegmentTupleIterator implements ITupleIterator {
closeTable();
}
+ @Override
+ public Range<Long> getCacheExcludedPeriod() {
+ return null;
+ }
+
private void closeScanner() {
if (logger.isDebugEnabled() && scan != null) {
logger.debug("Scan " + scan.toString());
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/hbase/SerializedHBaseTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/SerializedHBaseTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/hbase/SerializedHBaseTupleIterator.java
index 8c37347..cc30911 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/SerializedHBaseTupleIterator.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/SerializedHBaseTupleIterator.java
@@ -20,6 +20,7 @@ package org.apache.kylin.storage.hbase;
import java.util.*;
+import com.google.common.collect.Range;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeSegment;
@@ -135,4 +136,9 @@ public class SerializedHBaseTupleIterator implements ITupleIterator {
segmentIterator.close();
}
}
+
+ @Override
+ public Range<Long> getCacheExcludedPeriod() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/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
index 4ebe593..6eb539c 100644
--- 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
@@ -21,13 +21,17 @@ package org.apache.kylin.storage.hbase.coprocessor.endpoint;
import java.io.IOException;
import java.util.*;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.*;
import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.NotImplementedException;
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.DateFormat;
import org.apache.kylin.common.util.RangeUtil;
import org.apache.kylin.cube.kv.RowKeyColumnIO;
import org.apache.kylin.invertedindex.IISegment;
@@ -51,10 +55,6 @@ import org.apache.kylin.storage.tuple.TupleInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Range;
-import com.google.common.collect.Sets;
import com.google.protobuf.ByteString;
/**
@@ -80,11 +80,15 @@ public class EndpointTupleIterator implements ITupleIterator {
private final EndpointAggregators pushedDownAggregators;
private final Range<Long> tsRange;//timestamp column condition's interval
- Iterator<List<IIProtos.IIResponse.IIRow>> regionResponsesIterator = null;
- ITupleIterator tupleIterator = null;
- HTableInterface table = null;
+ private Iterator<List<IIProtos.IIResponse.IIRow>> regionResponsesIterator = null;
+ private ITupleIterator tupleIterator = null;
+ private HTableInterface table = null;
- int rowsInAllMetric = 0;
+ private TblColRef partitionCol;
+ private int shardCount;
+ private long currentShardLastDataTime = Long.MIN_VALUE;
+ private List<Long> shardLastDataTimes = Lists.newArrayList();
+ private int rowsInAllMetric = 0;
public EndpointTupleIterator(IISegment segment, TupleFilter rootFilter, Collection<TblColRef> groupBy, List<FunctionDesc> measures, StorageContext context, HConnection conn) throws Throwable {
@@ -128,7 +132,8 @@ public class EndpointTupleIterator implements ITupleIterator {
this.pushedDownAggregators = EndpointAggregators.fromFunctions(tableRecordInfo, measures);
int tsCol = this.tableRecordInfo.getTimestampColumn();
- this.tsRange = TsConditionExtractor.extractTsCondition(this.columns.get(tsCol), rootFilter);
+ 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");
@@ -138,9 +143,13 @@ public class EndpointTupleIterator implements ITupleIterator {
}
IIProtos.IIRequest endpointRequest = prepareRequest();
- regionResponsesIterator = getResults(endpointRequest, table);
+
+ Collection<List<IIProtos.IIResponse.IIRow>> shardResults = getResults(endpointRequest, table);
+ this.shardCount = shardResults.size();
+ this.regionResponsesIterator = shardResults.iterator();
if (this.regionResponsesIterator.hasNext()) {
+ this.currentShardLastDataTime = Long.MIN_VALUE;
this.tupleIterator = new SingleRegionTupleIterator(this.regionResponsesIterator.next());
} else {
this.tupleIterator = ITupleIterator.EMPTY_TUPLE_ITERATOR;
@@ -186,7 +195,10 @@ public class EndpointTupleIterator implements ITupleIterator {
@Override
public boolean hasNext() {
while (!this.tupleIterator.hasNext()) {
+ this.shardLastDataTimes.add(this.currentShardLastDataTime);
+
if (this.regionResponsesIterator.hasNext()) {
+ this.currentShardLastDataTime = Long.MIN_VALUE;
this.tupleIterator = new SingleRegionTupleIterator(this.regionResponsesIterator.next());
} else {
return false;
@@ -203,7 +215,11 @@ public class EndpointTupleIterator implements ITupleIterator {
throw new IllegalStateException("No more ITuple in EndpointTupleIterator");
}
- return this.tupleIterator.next();
+ ITuple tuple = this.tupleIterator.next();
+
+ //update shardLastDataTimes
+ this.currentShardLastDataTime = Tuple.getTs(tuple,this.partitionCol);
+ return tuple;
}
@Override
@@ -218,6 +234,13 @@ public class EndpointTupleIterator implements ITupleIterator {
logger.info("Closed after " + rowsInAllMetric + " rows are fetched");
}
+ @Override
+ public Range<Long> getCacheExcludedPeriod() {
+ Preconditions.checkArgument(shardCount == this.shardLastDataTimes.size());
+ long min = Collections.min(this.shardLastDataTimes);
+ return Ranges.atLeast(min);//inclusive
+ }
+
private IIProtos.IIRequest prepareRequest() throws IOException {
IIProtos.IIRequest.Builder builder = IIProtos.IIRequest.newBuilder();
@@ -237,7 +260,7 @@ public class EndpointTupleIterator implements ITupleIterator {
}
//TODO : async callback
- private Iterator<List<IIProtos.IIResponse.IIRow>> getResults(final IIProtos.IIRequest request, HTableInterface table) throws Throwable {
+ private Collection<List<IIProtos.IIResponse.IIRow>> getResults(final IIProtos.IIRequest request, HTableInterface table) throws Throwable {
Map<byte[], List<IIProtos.IIResponse.IIRow>> results = table.coprocessorService(IIProtos.RowsService.class, null, null, new Batch.Call<IIProtos.RowsService, List<IIProtos.IIResponse.IIRow>>() {
public List<IIProtos.IIResponse.IIRow> call(IIProtos.RowsService rowsService) throws IOException {
ServerRpcController controller = new ServerRpcController();
@@ -252,7 +275,7 @@ public class EndpointTupleIterator implements ITupleIterator {
}
});
- return results.values().iterator();
+ return results.values();
}
private TupleInfo buildTupleInfo() {
@@ -296,7 +319,6 @@ public class EndpointTupleIterator implements ITupleIterator {
/**
* Internal class to handle iterators for a single region's returned rows
- *
*/
class SingleRegionTupleIterator implements ITupleIterator {
private List<IIProtos.IIResponse.IIRow> rows;
@@ -350,6 +372,11 @@ public class EndpointTupleIterator implements ITupleIterator {
public void close() {
}
+ @Override
+ public Range<Long> getCacheExcludedPeriod() {
+ throw new NotImplementedException();
+ }
+
private ITuple makeTuple(TableRecord tableRecord, List<Object> measureValues) {
this.tuple = new Tuple(tupleInfo);
// groups
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
index abe6157..7ac9f2b 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/IIEndpoint.java
@@ -244,26 +244,6 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
return responseBuilder.build();
}
- private void decodeWithDictionary(byte[] recordBuffer, RawTableRecord encodedRecord, Dictionary<?>[] localDictionaries, TableRecordInfoDigest digest, RowKeyColumnIO rowKeyColumnIO, CoprocessorRowType coprocessorRowType) {
- final TblColRef[] columns = coprocessorRowType.columns;
- final int columnSize = columns.length;
- final boolean[] isMetric = digest.isMetrics();
- final boolean emptyDictionary = Array.isEmpty(localDictionaries);
- for (int i = 0; i < columnSize; i++) {
- final TblColRef column = columns[i];
- if (isMetric[i]) {
- rowKeyColumnIO.writeColumnWithoutDictionary(encodedRecord.getBytes(), encodedRecord.offset(i), encodedRecord.length(i), recordBuffer, digest.offset(i), rowKeyColumnIO.getColumnLength(column));
- } else {
- if (emptyDictionary) {
- rowKeyColumnIO.writeColumnWithoutDictionary(encodedRecord.getBytes(), encodedRecord.offset(i), encodedRecord.length(i), recordBuffer, digest.offset(i), rowKeyColumnIO.getColumnLength(column));
- } else {
- final Dictionary<?> localDictionary = localDictionaries[i];
- final byte[] valueBytesFromId = localDictionary.getValueBytesFromId(encodedRecord.getValueID(i));
- rowKeyColumnIO.writeColumnWithoutDictionary(valueBytesFromId, 0, valueBytesFromId.length, recordBuffer, digest.offset(i), rowKeyColumnIO.getColumnLength(column));
- }
- }
- }
- }
private IIProtos.IIResponse getNonAggregatedResponse(Iterable<Slice> slices, TableRecordInfoDigest recordInfo, CoprocessorFilter filter, CoprocessorRowType type) {
IIProtos.IIResponse.Builder responseBuilder = IIProtos.IIResponse.newBuilder();
@@ -299,6 +279,27 @@ public class IIEndpoint extends IIProtos.RowsService implements Coprocessor, Cop
return responseBuilder.build();
}
+ private void decodeWithDictionary(byte[] recordBuffer, RawTableRecord encodedRecord, Dictionary<?>[] localDictionaries, TableRecordInfoDigest digest, RowKeyColumnIO rowKeyColumnIO, CoprocessorRowType coprocessorRowType) {
+ final TblColRef[] columns = coprocessorRowType.columns;
+ final int columnSize = columns.length;
+ final boolean[] isMetric = digest.isMetrics();
+ final boolean emptyDictionary = Array.isEmpty(localDictionaries);
+ for (int i = 0; i < columnSize; i++) {
+ final TblColRef column = columns[i];
+ if (isMetric[i]) {
+ rowKeyColumnIO.writeColumnWithoutDictionary(encodedRecord.getBytes(), encodedRecord.offset(i), encodedRecord.length(i), recordBuffer, digest.offset(i), rowKeyColumnIO.getColumnLength(column));
+ } else {
+ if (emptyDictionary) {
+ rowKeyColumnIO.writeColumnWithoutDictionary(encodedRecord.getBytes(), encodedRecord.offset(i), encodedRecord.length(i), recordBuffer, digest.offset(i), rowKeyColumnIO.getColumnLength(column));
+ } else {
+ final Dictionary<?> localDictionary = localDictionaries[i];
+ final byte[] valueBytesFromId = localDictionary.getValueBytesFromId(encodedRecord.getValueID(i));
+ rowKeyColumnIO.writeColumnWithoutDictionary(valueBytesFromId, 0, valueBytesFromId.length, recordBuffer, digest.offset(i), rowKeyColumnIO.getColumnLength(column));
+ }
+ }
+ }
+ }
+
@Override
public void start(CoprocessorEnvironment env) throws IOException {
if (env instanceof RegionCoprocessorEnvironment) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/generated/IIProtos.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/generated/IIProtos.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/generated/IIProtos.java
index 3b87f08..9c3d4b5 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/generated/IIProtos.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/generated/IIProtos.java
@@ -876,16 +876,6 @@ public final class IIProtos {
*/
org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos.IIResponse.IIRowOrBuilder getRowsOrBuilder(
int index);
-
- // optional int64 lastFinalizedTime = 2;
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- boolean hasLastFinalizedTime();
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- long getLastFinalizedTime();
}
/**
* Protobuf type {@code IIResponse}
@@ -946,11 +936,6 @@ public final class IIProtos {
rows_.add(input.readMessage(org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos.IIResponse.IIRow.PARSER, extensionRegistry));
break;
}
- case 16: {
- bitField0_ |= 0x00000001;
- lastFinalizedTime_ = input.readInt64();
- break;
- }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -1531,7 +1516,6 @@ public final class IIProtos {
// @@protoc_insertion_point(class_scope:IIResponse.IIRow)
}
- private int bitField0_;
// repeated .IIResponse.IIRow rows = 1;
public static final int ROWS_FIELD_NUMBER = 1;
private java.util.List<org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos.IIResponse.IIRow> rows_;
@@ -1568,25 +1552,8 @@ public final class IIProtos {
return rows_.get(index);
}
- // optional int64 lastFinalizedTime = 2;
- public static final int LASTFINALIZEDTIME_FIELD_NUMBER = 2;
- private long lastFinalizedTime_;
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- public boolean hasLastFinalizedTime() {
- return ((bitField0_ & 0x00000001) == 0x00000001);
- }
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- public long getLastFinalizedTime() {
- return lastFinalizedTime_;
- }
-
private void initFields() {
rows_ = java.util.Collections.emptyList();
- lastFinalizedTime_ = 0L;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -1609,9 +1576,6 @@ public final class IIProtos {
for (int i = 0; i < rows_.size(); i++) {
output.writeMessage(1, rows_.get(i));
}
- if (((bitField0_ & 0x00000001) == 0x00000001)) {
- output.writeInt64(2, lastFinalizedTime_);
- }
getUnknownFields().writeTo(output);
}
@@ -1625,10 +1589,6 @@ public final class IIProtos {
size += com.google.protobuf.CodedOutputStream
.computeMessageSize(1, rows_.get(i));
}
- if (((bitField0_ & 0x00000001) == 0x00000001)) {
- size += com.google.protobuf.CodedOutputStream
- .computeInt64Size(2, lastFinalizedTime_);
- }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -1654,11 +1614,6 @@ public final class IIProtos {
boolean result = true;
result = result && getRowsList()
.equals(other.getRowsList());
- result = result && (hasLastFinalizedTime() == other.hasLastFinalizedTime());
- if (hasLastFinalizedTime()) {
- result = result && (getLastFinalizedTime()
- == other.getLastFinalizedTime());
- }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -1676,10 +1631,6 @@ public final class IIProtos {
hash = (37 * hash) + ROWS_FIELD_NUMBER;
hash = (53 * hash) + getRowsList().hashCode();
}
- if (hasLastFinalizedTime()) {
- hash = (37 * hash) + LASTFINALIZEDTIME_FIELD_NUMBER;
- hash = (53 * hash) + hashLong(getLastFinalizedTime());
- }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -1796,8 +1747,6 @@ public final class IIProtos {
} else {
rowsBuilder_.clear();
}
- lastFinalizedTime_ = 0L;
- bitField0_ = (bitField0_ & ~0x00000002);
return this;
}
@@ -1825,7 +1774,6 @@ public final class IIProtos {
public org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos.IIResponse buildPartial() {
org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos.IIResponse result = new org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.IIProtos.IIResponse(this);
int from_bitField0_ = bitField0_;
- int to_bitField0_ = 0;
if (rowsBuilder_ == null) {
if (((bitField0_ & 0x00000001) == 0x00000001)) {
rows_ = java.util.Collections.unmodifiableList(rows_);
@@ -1835,11 +1783,6 @@ public final class IIProtos {
} else {
result.rows_ = rowsBuilder_.build();
}
- if (((from_bitField0_ & 0x00000002) == 0x00000002)) {
- to_bitField0_ |= 0x00000001;
- }
- result.lastFinalizedTime_ = lastFinalizedTime_;
- result.bitField0_ = to_bitField0_;
onBuilt();
return result;
}
@@ -1881,9 +1824,6 @@ public final class IIProtos {
}
}
}
- if (other.hasLastFinalizedTime()) {
- setLastFinalizedTime(other.getLastFinalizedTime());
- }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -2157,39 +2097,6 @@ public final class IIProtos {
return rowsBuilder_;
}
- // optional int64 lastFinalizedTime = 2;
- private long lastFinalizedTime_ ;
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- public boolean hasLastFinalizedTime() {
- return ((bitField0_ & 0x00000002) == 0x00000002);
- }
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- public long getLastFinalizedTime() {
- return lastFinalizedTime_;
- }
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- public Builder setLastFinalizedTime(long value) {
- bitField0_ |= 0x00000002;
- lastFinalizedTime_ = value;
- onChanged();
- return this;
- }
- /**
- * <code>optional int64 lastFinalizedTime = 2;</code>
- */
- public Builder clearLastFinalizedTime() {
- bitField0_ = (bitField0_ & ~0x00000002);
- lastFinalizedTime_ = 0L;
- onChanged();
- return this;
- }
-
// @@protoc_insertion_point(builder_scope:IIResponse)
}
@@ -2461,14 +2368,13 @@ public final class IIProtos {
"/storage/hbase/coprocessor/endpoint/prot" +
"obuf/II.proto\"a\n\tIIRequest\022\014\n\004type\030\001 \002(\014" +
"\022\016\n\006filter\030\002 \002(\014\022\021\n\tprojector\030\003 \002(\014\022\022\n\na" +
- "ggregator\030\004 \002(\014\022\017\n\007tsRange\030\005 \001(\014\"t\n\nIIRe" +
- "sponse\022\037\n\004rows\030\001 \003(\0132\021.IIResponse.IIRow\022" +
- "\031\n\021lastFinalizedTime\030\002 \001(\003\032*\n\005IIRow\022\017\n\007c" +
- "olumns\030\001 \002(\014\022\020\n\010measures\030\002 \001(\01421\n\013RowsSe" +
- "rvice\022\"\n\007getRows\022\n.IIRequest\032\013.IIRespons" +
- "eBQ\n=org.apache.kylin.storage.hbase.copr",
- "ocessor.endpoint.generatedB\010IIProtosH\001\210\001" +
- "\001\240\001\001"
+ "ggregator\030\004 \002(\014\022\017\n\007tsRange\030\005 \001(\014\"Y\n\nIIRe" +
+ "sponse\022\037\n\004rows\030\001 \003(\0132\021.IIResponse.IIRow\032" +
+ "*\n\005IIRow\022\017\n\007columns\030\001 \002(\014\022\020\n\010measures\030\002 " +
+ "\001(\01421\n\013RowsService\022\"\n\007getRows\022\n.IIReques" +
+ "t\032\013.IIResponseBQ\n=org.apache.kylin.stora" +
+ "ge.hbase.coprocessor.endpoint.generatedB",
+ "\010IIProtosH\001\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -2486,7 +2392,7 @@ public final class IIProtos {
internal_static_IIResponse_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_IIResponse_descriptor,
- new java.lang.String[] { "Rows", "LastFinalizedTime", });
+ new java.lang.String[] { "Rows", });
internal_static_IIResponse_IIRow_descriptor =
internal_static_IIResponse_descriptor.getNestedTypes().get(0);
internal_static_IIResponse_IIRow_fieldAccessorTable = new
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/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
index e43bb6e..568dffc 100644
--- 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
@@ -22,7 +22,6 @@ message IIResponse {
optional bytes measures = 2;
}
repeated IIRow rows = 1;
- optional int64 lastFinalizedTime = 2;
}
service RowsService {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/fba32d88/storage/src/main/java/org/apache/kylin/storage/tuple/Tuple.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/tuple/Tuple.java b/storage/src/main/java/org/apache/kylin/storage/tuple/Tuple.java
index 97d447f..90c5664 100644
--- a/storage/src/main/java/org/apache/kylin/storage/tuple/Tuple.java
+++ b/storage/src/main/java/org/apache/kylin/storage/tuple/Tuple.java
@@ -129,9 +129,17 @@ public class Tuple implements ITuple {
return 1L * days * (1000 * 3600 * 24);
}
- public static int millisToEpicDays(long millis)
- {
- return (int)(millis / (1000 * 3600 * 24));
+ public static int millisToEpicDays(long millis) {
+ return (int) (millis / (1000 * 3600 * 24));
+ }
+
+ public static long getTs(ITuple row, TblColRef partitionCol) {
+ //ts column type differentiate
+ if (partitionCol.getDatatype().equals("date")) {
+ return Tuple.epicDaysToMillis(Integer.valueOf(row.getValue(partitionCol).toString()));
+ } else {
+ return Long.valueOf(row.getValue(partitionCol).toString());
+ }
}
public static Object convertOptiqCellValue(String strValue, String dataType) {