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:12 UTC
[30/50] incubator-kylin git commit: KYLIN-671 cachefledgedtupleiteator
KYLIN-671 cachefledgedtupleiteator
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/dea04bf1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/dea04bf1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/dea04bf1
Branch: refs/heads/streaming-localdict
Commit: dea04bf1ffa24d0a936840c5be5b5ce690e02383
Parents: 7568872
Author: honma <ho...@ebay.com>
Authored: Wed Apr 15 10:40:16 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Wed Apr 22 16:45:38 2015 +0800
----------------------------------------------------------------------
.../org/apache/kylin/common/util/RangeUtil.java | 24 ++++
.../org/apache/kylin/common/util/BasicTest.java | 18 ++-
.../org/apache/kylin/common/util/RangeTest.java | 26 +++++
.../org/apache/kylin/common/util/RangeUtil.java | 18 ---
.../metadata/realization/SQLDigestUtil.java | 113 +++++++++++++++++++
.../metadata/tuple/CompoundTupleIterator.java | 42 +++++++
.../kylin/metadata/tuple/ITupleIterator.java | 19 ++--
.../metadata/tuple/SimpleTupleIterator.java | 35 ++++++
pom.xml | 2 +-
.../kylin/query/enumerator/OLAPEnumerator.java | 6 +-
.../kylin/storage/StorageEngineFactory.java | 7 +-
.../cache/CacheFledgedTupleIterator.java | 62 ++++++++--
.../kylin/storage/cache/StreamSQLResult.java | 23 +++-
.../cube/SerializedCubeTupleIterator.java | 5 +
.../storage/hbase/CubeSegmentTupleIterator.java | 6 +
.../hbase/SerializedHBaseTupleIterator.java | 18 ++-
.../endpoint/EndpointTupleIterator.java | 18 ++-
.../endpoint/TsConditionExtractor.java | 20 ++--
.../storage/hybrid/HybridStorageEngine.java | 83 ++++----------
.../storage/hybrid/HybridTupleIterator.java | 52 ---------
.../endpoint/TsConditionExtractorTest.java | 7 +-
.../apache/kylin/storage/test/StorageTest.java | 7 +-
.../invertedindex/IIStreamBuilder.java | 4 +-
23 files changed, 417 insertions(+), 198 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/common/src/main/java/org/apache/kylin/common/util/RangeUtil.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/kylin/common/util/RangeUtil.java b/common/src/main/java/org/apache/kylin/common/util/RangeUtil.java
new file mode 100644
index 0000000..fae521c
--- /dev/null
+++ b/common/src/main/java/org/apache/kylin/common/util/RangeUtil.java
@@ -0,0 +1,24 @@
+package org.apache.kylin.common.util;
+
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Range;
+import com.google.common.collect.RangeSet;
+import com.google.common.collect.TreeRangeSet;
+
+/**
+ * Created by Hongbin Ma(Binmahone) on 4/14/15.
+ */
+public class RangeUtil {
+ /**
+ * remove from self the elements that exist in other
+ * @return
+ */
+ public static <C extends Comparable<?>> List<Range<C>> remove(Range<C> self, Range<C> other) {
+ RangeSet<C> rangeSet = TreeRangeSet.create();
+ rangeSet.add(self);
+ rangeSet.remove(other);
+ return Lists.newArrayList(rangeSet.asRanges());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/common/src/test/java/org/apache/kylin/common/util/BasicTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/kylin/common/util/BasicTest.java b/common/src/test/java/org/apache/kylin/common/util/BasicTest.java
index a949599..a178311 100644
--- a/common/src/test/java/org/apache/kylin/common/util/BasicTest.java
+++ b/common/src/test/java/org/apache/kylin/common/util/BasicTest.java
@@ -22,21 +22,17 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Calendar;
-import com.google.common.base.Function;
-import com.google.common.collect.Lists;
-
-import com.google.common.collect.Ordering;
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.lang.math.IntRange;
-import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Range;
+
/**
* Created by honma on 10/17/14.
* <p/>
@@ -70,8 +66,8 @@ public class BasicTest {
public void test1() throws Exception {
- System.out.println(Ranges.open(3, 5).isConnected(Ranges.open(4, 10)));
- System.out.println(Ranges.open(4, 10).isConnected(Ranges.open(3,5)));
+ System.out.println(Range.open(3, 5).isConnected(Range.open(4, 10)));
+ System.out.println(Range.open(4, 10).isConnected(Range.open(3,5)));
String bb = "\\x00\\x00\\x00\\x00\\x01\\x3F\\xD0\\x2D\\58\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00";//2013/07/12 07:59:37
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/common/src/test/java/org/apache/kylin/common/util/RangeTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/kylin/common/util/RangeTest.java b/common/src/test/java/org/apache/kylin/common/util/RangeTest.java
new file mode 100644
index 0000000..59eb9d7
--- /dev/null
+++ b/common/src/test/java/org/apache/kylin/common/util/RangeTest.java
@@ -0,0 +1,26 @@
+package org.apache.kylin.common.util;
+
+import com.google.common.collect.Range;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Created by Hongbin Ma(Binmahone) on 4/14/15.
+ */
+public class RangeTest {
+ @Test
+ public void basicTest() {
+ Range a0 = Range.closed(2, 5);
+ Range a1 = Range.open(2, 5);
+ Range a2 = Range.open(2, 5);
+ Range a3 = Range.open(2, 3);
+ Range a4 = Range.open(2, 9);
+
+ Assert.assertTrue(RangeUtil.remove(a0, a1).size() == 2);
+ Assert.assertTrue(RangeUtil.remove(a1, a2).size() == 0);
+ Assert.assertTrue(RangeUtil.remove(a1, a3).size() == 1);
+ Assert.assertTrue(RangeUtil.remove(a1, a3).get(0).equals(Range.closedOpen(3, 5)));
+ Assert.assertTrue(RangeUtil.remove(a1, a4).size() == 0);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/common/src/test/java/org/apache/kylin/common/util/RangeUtil.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/kylin/common/util/RangeUtil.java b/common/src/test/java/org/apache/kylin/common/util/RangeUtil.java
deleted file mode 100644
index 2b74b32..0000000
--- a/common/src/test/java/org/apache/kylin/common/util/RangeUtil.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package org.apache.kylin.common.util;
-
-import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
-
-/**
- * Created by Hongbin Ma(Binmahone) on 4/14/15.
- */
-public class RangeUtil {
- public static <C extends Comparable<?>> Range<C> retain(Range<C> self, Range<C> other) {
- if (!self.isConnected(other) || self.intersection(other).isEmpty()) {
- return null;
- }
- return null;
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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
new file mode 100644
index 0000000..f1861b0
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kylin/metadata/realization/SQLDigestUtil.java
@@ -0,0 +1,113 @@
+package org.apache.kylin.metadata.realization;
+
+import com.google.common.base.Function;
+import com.google.common.collect.BoundType;
+import com.google.common.collect.Range;
+import org.apache.kylin.metadata.filter.*;
+import org.apache.kylin.metadata.model.DataType;
+import org.apache.kylin.metadata.model.TblColRef;
+import org.apache.kylin.metadata.util.DateFormat;
+
+/**
+ * Created by Hongbin Ma(Binmahone) on 4/14/15.
+ */
+public class SQLDigestUtil {
+
+ public static <F, T> T appendTsFilterToExecute(SQLDigest sqlDigest, TblColRef partitionColRef, Range<Long> tsRange, Function<F, T> action) {
+ // add the boundary condition to query real-time
+ TupleFilter originalFilter = sqlDigest.filter;
+ sqlDigest.filter = createFilterForRealtime(originalFilter, partitionColRef, tsRange);
+
+ boolean addFilterColumn = false, addAllColumn = false;
+
+ if (!sqlDigest.filterColumns.contains(partitionColRef)) {
+ sqlDigest.filterColumns.add(partitionColRef);
+ addFilterColumn = true;
+ }
+
+ if (!sqlDigest.allColumns.contains(partitionColRef)) {
+ sqlDigest.allColumns.add(partitionColRef);
+ addAllColumn = true;
+ }
+
+ T ret = action.apply(null);
+
+ // restore the sqlDigest
+ sqlDigest.filter = originalFilter;
+
+ if (addFilterColumn)
+ sqlDigest.filterColumns.remove(partitionColRef);
+
+ if (addAllColumn)
+ sqlDigest.allColumns.remove(partitionColRef);
+
+ return ret;
+ }
+
+ private static String formatTimeStr(DataType type, long ts) {
+ String ret;
+ if (type == DataType.getInstance("date")) {
+ ret = DateFormat.formatToDateStr(ts);
+ } else if (type == DataType.getInstance("long")) {
+ ret = String.valueOf(ts);
+ } else {
+ throw new IllegalArgumentException("Illegal type for partition column " + type);
+ }
+ return ret;
+ }
+
+ private static TupleFilter createFilterForRealtime(TupleFilter originFilter, TblColRef partitionColRef, Range<Long> tsRange) {
+ DataType type = partitionColRef.getColumn().getType();
+
+ String startTimeStr, endTimeStr;
+ CompareTupleFilter startFilter = null, endFilter = null;
+ if (tsRange.hasLowerBound()) {
+ startTimeStr = formatTimeStr(type, tsRange.lowerEndpoint());
+ if (tsRange.lowerBoundType() == BoundType.CLOSED) {
+ startFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.GTE);
+ } else {
+ startFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.GT);
+ }
+ ColumnTupleFilter columnTupleFilter = new ColumnTupleFilter(partitionColRef);
+ ConstantTupleFilter constantTupleFilter = new ConstantTupleFilter(startTimeStr);
+ startFilter.addChild(columnTupleFilter);
+ startFilter.addChild(constantTupleFilter);
+ }
+
+ if (tsRange.hasUpperBound()) {
+ endTimeStr = formatTimeStr(type, tsRange.upperEndpoint());
+ if (tsRange.upperBoundType() == BoundType.CLOSED) {
+ endFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.LTE);
+ } else {
+ endFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.LT);
+ }
+ ColumnTupleFilter columnTupleFilter = new ColumnTupleFilter(partitionColRef);
+ ConstantTupleFilter constantTupleFilter = new ConstantTupleFilter(endTimeStr);
+ endFilter.addChild(columnTupleFilter);
+ endFilter.addChild(constantTupleFilter);
+ }
+
+ if (originFilter == null) {
+ if (endFilter == null) {
+ return startFilter;
+ }
+ if (startFilter == null) {
+ return endFilter;
+ }
+ }
+
+ LogicalTupleFilter logicalTupleFilter = new LogicalTupleFilter(TupleFilter.FilterOperatorEnum.AND);
+
+ if (originFilter != null) {
+ logicalTupleFilter.addChild(originFilter);
+ }
+ if (startFilter != null) {
+ logicalTupleFilter.addChild(startFilter);
+ }
+ if (endFilter != null) {
+ logicalTupleFilter.addChild(endFilter);
+ }
+
+ return logicalTupleFilter;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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
new file mode 100644
index 0000000..08562ca
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kylin/metadata/tuple/CompoundTupleIterator.java
@@ -0,0 +1,42 @@
+package org.apache.kylin.metadata.tuple;
+
+import com.google.common.collect.Iterators;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Created by Hongbin Ma(Binmahone) on 4/14/15.
+ */
+public class CompoundTupleIterator implements ITupleIterator {
+ private List<ITupleIterator> backends;
+ private Iterator<ITuple> compoundIterator;
+
+ public CompoundTupleIterator(List<ITupleIterator> backends) {
+ this.backends = backends;
+ this.compoundIterator = Iterators.concat(backends.iterator());
+ }
+
+ @Override
+ public void close() throws IOException {
+ for (ITupleIterator i : backends) {
+ i.close();
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.compoundIterator.hasNext();
+ }
+
+ @Override
+ public ITuple next() {
+ return this.compoundIterator.next();
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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 6966751..5ba8d9e 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
@@ -18,10 +18,14 @@
package org.apache.kylin.metadata.tuple;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
+
/**
* @author xjiang
*/
-public interface ITupleIterator {
+public interface ITupleIterator extends Iterator<ITuple>, Closeable {
public static final ITupleIterator EMPTY_TUPLE_ITERATOR = new ITupleIterator() {
@Override
public boolean hasNext() {
@@ -34,14 +38,15 @@ public interface ITupleIterator {
}
@Override
+ public void remove() {
+
+ }
+
+ @Override
public void close() {
}
};
- public boolean hasNext();
-
- public ITuple next();
-
- public void close();
-
+ @Override
+ public void close() throws IOException;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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
new file mode 100644
index 0000000..b3f0caa
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kylin/metadata/tuple/SimpleTupleIterator.java
@@ -0,0 +1,35 @@
+package org.apache.kylin.metadata.tuple;
+
+import java.util.Iterator;
+
+/**
+ * Created by Hongbin Ma(Binmahone) on 4/14/15.
+ *
+ */
+public class SimpleTupleIterator implements ITupleIterator {
+
+ private Iterator<ITuple> backend;
+
+ public SimpleTupleIterator(Iterator<ITuple> backend) {
+ this.backend = backend;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return backend.hasNext();
+ }
+
+ @Override
+ public ITuple next() {
+ return backend.next();
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index e750154..6b36e1e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -73,7 +73,7 @@
<log4j.version>1.2.17</log4j.version>
<slf4j.version>1.6.4</slf4j.version>
<jackson.version>2.2.3</jackson.version>
- <guava.version>12.0</guava.version>
+ <guava.version>18.0</guava.version>
<jsch.version>0.1.51</jsch.version>
<xerces.version>2.9.1</xerces.version>
<xalan.version>2.7.1</xalan.version>
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
----------------------------------------------------------------------
diff --git a/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java b/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
index f49853f..0300d70 100644
--- a/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
+++ b/query/src/main/java/org/apache/kylin/query/enumerator/OLAPEnumerator.java
@@ -90,9 +90,7 @@ public class OLAPEnumerator implements Enumerator<Object[]> {
@Override
public void close() {
- if (cursor != null) {
- cursor.close();
- }
+ IOUtils.closeQuietly(cursor);
}
private Object[] convertCurrentRow(ITuple tuple) {
@@ -138,7 +136,7 @@ public class OLAPEnumerator implements Enumerator<Object[]> {
bindVariable(olapContext.filter);
// query storage engine
- IStorageEngine storageEngine = StorageEngineFactory.getStorageEngine(olapContext.realization);
+ IStorageEngine storageEngine = StorageEngineFactory.getStorageEngine(olapContext.realization, true);
ITupleIterator iterator = storageEngine.search(olapContext.storageContext, olapContext.getSQLDigest());
if (logger.isDebugEnabled()) {
logger.debug("return TupleIterator...");
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/storage/src/main/java/org/apache/kylin/storage/StorageEngineFactory.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/StorageEngineFactory.java b/storage/src/main/java/org/apache/kylin/storage/StorageEngineFactory.java
index b2e82a8..df3371f 100644
--- a/storage/src/main/java/org/apache/kylin/storage/StorageEngineFactory.java
+++ b/storage/src/main/java/org/apache/kylin/storage/StorageEngineFactory.java
@@ -22,6 +22,7 @@ import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.invertedindex.IIInstance;
import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.RealizationType;
+import org.apache.kylin.storage.cache.CacheFledgedTupleIterator;
import org.apache.kylin.storage.hbase.CubeStorageEngine;
import org.apache.kylin.storage.hbase.InvertedIndexStorageEngine;
import org.apache.kylin.storage.hybrid.HybridInstance;
@@ -32,7 +33,11 @@ import org.apache.kylin.storage.hybrid.HybridStorageEngine;
*/
public class StorageEngineFactory {
- public static IStorageEngine getStorageEngine(IRealization realization) {
+ public static IStorageEngine getStorageEngine(IRealization realization, boolean allowCache) {
+ if (allowCache) {
+ return new CacheFledgedTupleIterator(realization);
+ }
+
if (realization.getType() == RealizationType.INVERTED_INDEX) {
return new InvertedIndexStorageEngine((IIInstance) realization);
} else if (realization.getType() == RealizationType.CUBE) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedTupleIterator.java
index 2b4c549..ab7ab3b 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedTupleIterator.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cache/CacheFledgedTupleIterator.java
@@ -1,28 +1,34 @@
package org.apache.kylin.storage.cache;
-import com.google.common.base.Preconditions;
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
import net.sf.ehcache.Cache;
import net.sf.ehcache.CacheManager;
import net.sf.ehcache.config.CacheConfiguration;
import net.sf.ehcache.config.MemoryUnit;
import net.sf.ehcache.config.PersistenceConfiguration;
import net.sf.ehcache.store.MemoryStoreEvictionPolicy;
+
import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.cube.CubeInstance;
+import org.apache.kylin.common.util.RangeUtil;
import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.model.DataModelDesc;
-import org.apache.kylin.metadata.model.ParameterDesc;
import org.apache.kylin.metadata.model.PartitionDesc;
import org.apache.kylin.metadata.model.TblColRef;
-import org.apache.kylin.metadata.realization.IRealization;
-import org.apache.kylin.metadata.realization.RealizationType;
-import org.apache.kylin.metadata.realization.SQLDigest;
-import org.apache.kylin.metadata.realization.StreamSQLDigest;
+import org.apache.kylin.metadata.realization.*;
+import org.apache.kylin.metadata.tuple.CompoundTupleIterator;
import org.apache.kylin.metadata.tuple.ITupleIterator;
+import org.apache.kylin.metadata.tuple.SimpleTupleIterator;
import org.apache.kylin.storage.IStorageEngine;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.StorageEngineFactory;
-import org.apache.kylin.storage.hbase.CubeSegmentTupleIterator;
+import org.apache.kylin.storage.hbase.coprocessor.endpoint.TsConditionExtractor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Range;
+
+import javax.annotation.Nullable;
+import java.util.List;
/**
* Created by Hongbin Ma(Binmahone) on 4/13/15.
@@ -30,7 +36,7 @@ import org.apache.kylin.storage.hbase.CubeSegmentTupleIterator;
public class CacheFledgedTupleIterator implements IStorageEngine {
public static final String SUCCESS_QUERY_CACHE = "SuccessQueryCache";
- public static final String EXCEPTION_QUERY_CACHE = "ExceptionQueryCache";
+ public static final String EXCEPTION_QUERY_CACHE = "ExceptionQueryCache";//TODO
static CacheManager cacheManager;
@@ -50,8 +56,11 @@ public class CacheFledgedTupleIterator implements IStorageEngine {
}
private TblColRef partitionColRef;
+ private IRealization realization;
public CacheFledgedTupleIterator(IRealization realization) {
+ this.realization = realization;
+
Preconditions.checkArgument(realization.getType() != RealizationType.CUBE, "Cube realization does not need dynamic cache!");
String modelName = realization.getModelName();
DataModelDesc dataModelDesc = MetadataManager.getInstance(KylinConfig.getInstanceFromEnv()).getDataModelDesc(modelName);
@@ -63,9 +72,38 @@ public class CacheFledgedTupleIterator implements IStorageEngine {
}
@Override
- public ITupleIterator search(StorageContext context, SQLDigest sqlDigest) {
+ public ITupleIterator search(final StorageContext context, final SQLDigest sqlDigest) {
StreamSQLDigest streamSQLDigest = new StreamSQLDigest(sqlDigest, partitionColRef);
- cacheManager.getCache(SUCCESS_QUERY_CACHE).get(streamSQLDigest).getObjectValue();
- return null;
+ StreamSQLResult cachedResult = (StreamSQLResult) cacheManager.getCache(SUCCESS_QUERY_CACHE).get(streamSQLDigest).getObjectValue();
+
+ ITupleIterator ret;
+
+ if (cachedResult != null) {
+ Range<Long> tsRange = TsConditionExtractor.extractTsCondition(partitionColRef, sqlDigest.filter);
+ Range<Long> reusePeriod = cachedResult.getReusableResults(tsRange);
+ if (reusePeriod != null) {
+ List<Range<Long>> remainings = RangeUtil.remove(tsRange, reusePeriod);
+ if (remainings.size() == 1) {
+ SimpleTupleIterator reusedTuples = new SimpleTupleIterator(cachedResult.reuse(reusePeriod));
+ Range<Long> remaining = remainings.get(0);
+ ITupleIterator freshTuples = SQLDigestUtil.appendTsFilterToExecute(sqlDigest, partitionColRef, remaining, new Function<Void, ITupleIterator>() {
+ @Nullable
+ @Override
+ public ITupleIterator apply(Void input) {
+ return StorageEngineFactory.getStorageEngine(realization, false).search(context, sqlDigest);
+ }
+ });
+ ret = new CompoundTupleIterator(Lists.newArrayList(reusedTuples, freshTuples));
+ //TODO:update cache
+ return ret;//cache successfully reused
+ }
+ }
+ }
+
+ //cache cannot reuse case:
+ ret = StorageEngineFactory.getStorageEngine(realization, false).search(context, sqlDigest);
+ //TODO:update cache
+ return ret;
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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 4d49032..b30db25 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
@@ -1,13 +1,34 @@
package org.apache.kylin.storage.cache;
+import com.google.common.collect.Range;
+import org.apache.kylin.metadata.tuple.ITuple;
import org.apache.kylin.storage.tuple.Tuple;
+import java.util.Iterator;
import java.util.List;
/**
* Created by Hongbin Ma(Binmahone) on 4/13/15.
*/
public class StreamSQLResult {
- List<Tuple> rows;
+ private List<ITuple> rows;
+ private Range<Long> timeCovered;
+ public Range<Long> getReusableResults(Range<Long> tsRange) {
+ if (tsRange.equals(timeCovered))
+ return timeCovered;
+
+ if (timeCovered.isConnected(tsRange)) {
+ //share nothing in common
+ return null;
+ }
+
+ Range<Long> ret = timeCovered.intersection(tsRange);
+ return ret.isEmpty() ? null : ret;
+ }
+
+ public Iterator<ITuple> reuse(Range<Long> reusablePeriod) {
+ //TODO: currently regardless of reusablePeriod, all rows are returned
+ return rows.iterator();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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 ea06b2a..e0605b4 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
@@ -24,6 +24,11 @@ public class SerializedCubeTupleIterator implements ITupleIterator {
}
@Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
public void close() {
// TODO Auto-generated method stub
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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 4190da8..e19db9e 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
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import org.apache.commons.lang.NotImplementedException;
import org.apache.kylin.storage.hbase.coprocessor.observer.ObserverEnabler;
import org.apache.hadoop.hbase.client.HConnection;
@@ -180,6 +181,11 @@ public class CubeSegmentTupleIterator implements ITupleIterator {
return this.tuple;
}
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
private void scanNextRange() {
if (this.rangeIterator.hasNext()) {
closeScanner();
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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 8a22579..7e8892d 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
@@ -18,6 +18,7 @@
package org.apache.kylin.storage.hbase;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
@@ -106,7 +107,9 @@ public class SerializedHBaseTupleIterator implements ITupleIterator {
}
@Override
- public ITuple next() {
+ public ITuple next() {
+ //TODO: last not closed
+
ITuple t = null;
while (hasNext()) {
if (segmentIterator.hasNext()) {
@@ -114,7 +117,11 @@ public class SerializedHBaseTupleIterator implements ITupleIterator {
scanCount++;
break;
} else {
- segmentIterator.close();
+ try {
+ segmentIterator.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
segmentIterator = segmentIteratorIterator.next();
}
}
@@ -122,7 +129,12 @@ public class SerializedHBaseTupleIterator implements ITupleIterator {
}
@Override
- public void close() {
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() throws IOException {
context.setTotalScanCount(scanCount);
segmentIterator.close();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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 1d765b5..5461de3 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
@@ -126,7 +126,8 @@ public class EndpointTupleIterator implements ITupleIterator {
this.pushedDownProjector = CoprocessorProjector.makeForEndpoint(tableRecordInfo, groupBy);
this.pushedDownAggregators = EndpointAggregators.fromFunctions(tableRecordInfo, measures);
- this.tsRange = TsConditionExtractor.extractTsCondition(tableRecordInfo, this.columns, rootFilter);
+ int tsCol = this.tableRecordInfo.getTimestampColumn();
+ this.tsRange = TsConditionExtractor.extractTsCondition(this.columns.get(tsCol), rootFilter);
IIProtos.IIRequest endpointRequest = prepareRequest();
regionResponsesIterator = getResults(endpointRequest, table);
@@ -198,7 +199,13 @@ public class EndpointTupleIterator implements ITupleIterator {
}
@Override
- public void close() {
+ public void remove() {
+ throw new UnsupportedOperationException();
+
+ }
+
+ @Override
+ public void close() throws IOException {
IOUtils.closeQuietly(table);
logger.info("Closed after " + rowsInAllMetric + " rows are fetched");
}
@@ -326,10 +333,15 @@ public class EndpointTupleIterator implements ITupleIterator {
}
@Override
- public void close() {
+ public void remove() {
+ throw new UnsupportedOperationException();
}
+ @Override
+ public void close() {
+ }
+
private ITuple makeTuple(TableRecord tableRecord, List<Object> measureValues) {
// groups
List<String> columnValues = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java
index bedf9f8..53f2945 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractor.java
@@ -1,8 +1,5 @@
package org.apache.kylin.storage.hbase.coprocessor.endpoint;
-import java.util.List;
-
-import org.apache.kylin.invertedindex.index.TableRecordInfo;
import org.apache.kylin.metadata.filter.CompareTupleFilter;
import org.apache.kylin.metadata.filter.LogicalTupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter;
@@ -10,16 +7,13 @@ import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.util.DateFormat;
import com.google.common.collect.Range;
-import com.google.common.collect.Ranges;
/**
* Created by Hongbin Ma(Binmahone) on 4/8/15.
*/
public class TsConditionExtractor {
- public static Range<Long> extractTsCondition(TableRecordInfo recordInfo, List<TblColRef> columns, TupleFilter rootFilter) {
- int tsCol = recordInfo.getTimestampColumn();
- TblColRef tsColRef = columns.get(tsCol);
+ public static Range<Long> extractTsCondition(TblColRef tsColRef, TupleFilter rootFilter) {
return extractTsConditionInternal(rootFilter, tsColRef);
}
@@ -27,7 +21,7 @@ public class TsConditionExtractor {
if (filter instanceof LogicalTupleFilter) {
if (filter.getOperator() == TupleFilter.FilterOperatorEnum.AND) {
- Range ret = Ranges.all();
+ Range ret = Range.all();
for (TupleFilter child : filter.getChildren()) {
Range childRange = extractTsConditionInternal(child, colRef);
if (childRange != null) {
@@ -51,19 +45,19 @@ public class TsConditionExtractor {
switch (compareTupleFilter.getOperator()) {
case EQ:
t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.closed(t, t);
+ return Range.closed(t, t);
case LT:
t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.lessThan(t);
+ return Range.lessThan(t);
case LTE:
t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.atMost(t);
+ return Range.atMost(t);
case GT:
t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.greaterThan(t);
+ return Range.greaterThan(t);
case GTE:
t = DateFormat.stringToMillis((String) firstValue);
- return Ranges.atLeast(t);
+ return Range.atLeast(t);
case NEQ:
case IN://not handled for now
break;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/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
index 7274fd6..3ac77b3 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridStorageEngine.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridStorageEngine.java
@@ -1,21 +1,25 @@
package org.apache.kylin.storage.hybrid;
+import com.google.common.base.Function;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Range;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.metadata.MetadataManager;
-import org.apache.kylin.metadata.filter.*;
import org.apache.kylin.metadata.model.DataModelDesc;
-import org.apache.kylin.metadata.model.DataType;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.SQLDigest;
+import org.apache.kylin.metadata.realization.SQLDigestUtil;
+import org.apache.kylin.metadata.tuple.CompoundTupleIterator;
import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.apache.kylin.metadata.util.DateFormat;
import org.apache.kylin.storage.IStorageEngine;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.StorageEngineFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
/**
* Created by shaoshi on 2/13/15.
*/
@@ -30,7 +34,7 @@ public class HybridStorageEngine implements IStorageEngine {
}
@Override
- public ITupleIterator search(StorageContext context, SQLDigest sqlDigest) {
+ public ITupleIterator search(final StorageContext context, final SQLDigest sqlDigest) {
// search the historic realization
ITupleIterator iterator1 = searchRealization(hybridInstance.getHistoryRealizationInstance(), context, sqlDigest);
@@ -44,72 +48,25 @@ public class HybridStorageEngine implements IStorageEngine {
return iterator1;
TblColRef partitionColRef = modelDesc.getPartitionDesc().getPartitionDateColumnRef();
- DataType partitionColType = partitionColRef.getColumn().getType();
-
- // add the boundary condition to query real-time
-
- TupleFilter originalFilter = sqlDigest.filter;
- sqlDigest.filter = createFilterForRealtime(originalFilter, partitionColRef, partitionColType, hybridInstance.getHistoryRealizationInstance().getDateRangeEnd());
-
- boolean addFilterColumn = false, addAllColumn = false;
-
- if (!sqlDigest.filterColumns.contains(partitionColRef)) {
- sqlDigest.filterColumns.add(partitionColRef);
- addFilterColumn = true;
- }
-
- if (!sqlDigest.allColumns.contains(partitionColRef)) {
- sqlDigest.allColumns.add(partitionColRef);
- addAllColumn = true;
- }
-
- // query real-time
- ITupleIterator iterator2 = searchRealization(hybridInstance.getRealTimeRealizationInstance(), context, sqlDigest);
-
- // restore the sqlDigest
- sqlDigest.filter = originalFilter;
-
- if (addFilterColumn)
- sqlDigest.filterColumns.remove(partitionColRef);
- if (addAllColumn)
- sqlDigest.allColumns.remove(partitionColRef);
+ ITupleIterator iterator2 = SQLDigestUtil.appendTsFilterToExecute(sqlDigest, partitionColRef, //
+ Range.atLeast(hybridInstance.getHistoryRealizationInstance().getDateRangeEnd()),//
+ new Function<Void, ITupleIterator>() {
+ @Nullable
+ @Override
+ public ITupleIterator apply(@Nullable Void input) {
+ ITupleIterator iterator2 = searchRealization(hybridInstance.getRealTimeRealizationInstance(), context, sqlDigest);
+ return iterator2;
+ }
+ });
// combine history and real-time tuple iterator
- return new HybridTupleIterator(new ITupleIterator[] { iterator1, iterator2 });
- }
-
- private TupleFilter createFilterForRealtime(TupleFilter originFilter, TblColRef partitionColRef, DataType type, long startDate) {
-
- String boundaryDate;
- if (type == DataType.getInstance("date")) {
- boundaryDate = DateFormat.formatToDateStr(startDate);
- } else if (type == DataType.getInstance("long")) {
- boundaryDate = String.valueOf(startDate);
- } else {
- throw new IllegalArgumentException("Illegal type for partition column " + type);
- }
-
- CompareTupleFilter compareTupleFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.GTE);
- ColumnTupleFilter columnTupleFilter = new ColumnTupleFilter(partitionColRef);
- ConstantTupleFilter constantTupleFilter = new ConstantTupleFilter(boundaryDate);
- compareTupleFilter.addChild(columnTupleFilter);
- compareTupleFilter.addChild(constantTupleFilter);
-
- if (originFilter == null)
- return compareTupleFilter;
-
- LogicalTupleFilter logicalTupleFilter = new LogicalTupleFilter(TupleFilter.FilterOperatorEnum.AND);
-
- logicalTupleFilter.addChild(originFilter);
- logicalTupleFilter.addChild(compareTupleFilter);
-
- return logicalTupleFilter;
+ return new CompoundTupleIterator(Lists.newArrayList(iterator1, iterator2));
}
private ITupleIterator searchRealization(IRealization realization, StorageContext context, SQLDigest sqlDigest) {
- IStorageEngine storageEngine = StorageEngineFactory.getStorageEngine(realization);
+ IStorageEngine storageEngine = StorageEngineFactory.getStorageEngine(realization,false);
return storageEngine.search(context, sqlDigest);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridTupleIterator.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridTupleIterator.java b/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridTupleIterator.java
deleted file mode 100644
index 99737ee..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/hybrid/HybridTupleIterator.java
+++ /dev/null
@@ -1,52 +0,0 @@
-package org.apache.kylin.storage.hybrid;
-
-import org.apache.kylin.metadata.tuple.ITuple;
-import org.apache.kylin.metadata.tuple.ITupleIterator;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Created by shaoshi on 2/27/15.
- */
-public class HybridTupleIterator implements ITupleIterator {
-
- private final static Logger logger = LoggerFactory.getLogger(HybridTupleIterator.class);
-
- private ITupleIterator[] iterators;
-
- private int currentIndex;
-
- public HybridTupleIterator(ITupleIterator[] iterators) {
- this.iterators = iterators;
- currentIndex = 0;
- }
-
- @Override
- public boolean hasNext() {
- if (iterators[currentIndex].hasNext())
- return true;
-
- while (!iterators[currentIndex].hasNext() && currentIndex + 1 < iterators.length) {
- currentIndex++;
- }
-
- return iterators[currentIndex].hasNext();
- }
-
- @Override
- public ITuple next() {
- while (!iterators[currentIndex].hasNext() && currentIndex + 1 < iterators.length) {
- currentIndex++;
- }
-
- return iterators[currentIndex].next();
- }
-
- @Override
- public void close() {
- logger.info("Closing HybridTupleIterator");
- for (ITupleIterator i : iterators) {
- i.close();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/storage/src/test/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractorTest.java
----------------------------------------------------------------------
diff --git a/storage/src/test/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractorTest.java b/storage/src/test/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractorTest.java
index d126d02..a5b5d09 100644
--- a/storage/src/test/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractorTest.java
+++ b/storage/src/test/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/TsConditionExtractorTest.java
@@ -70,7 +70,8 @@ public class TsConditionExtractorTest extends LocalFileMetadataTestCase {
CompareTupleFilter aFilter = new CompareTupleFilter(TupleFilter.FilterOperatorEnum.GT);
aFilter.addChild(new ColumnTupleFilter(calDt));
aFilter.addChild(new ConstantTupleFilter("2000-01-01"));
- Range<Long> range = TsConditionExtractor.extractTsCondition(tableRecordInfo, ii.getAllColumns(), aFilter);
+
+ Range<Long> range = TsConditionExtractor.extractTsCondition(ii.getAllColumns().get(tableRecordInfo.getTimestampColumn()), aFilter);
Assert.assertEquals(946684800000L, range.lowerEndpoint().longValue());
Assert.assertEquals(BoundType.OPEN, range.lowerBoundType());
Assert.assertTrue(!range.hasUpperBound());
@@ -97,7 +98,7 @@ public class TsConditionExtractorTest extends LocalFileMetadataTestCase {
LogicalTupleFilter rootFilter = new LogicalTupleFilter(TupleFilter.FilterOperatorEnum.AND);
rootFilter.addChildren(Lists.newArrayList(aFilter, bFilter, cFilter, dFilter));
- Range<Long> range = TsConditionExtractor.extractTsCondition(tableRecordInfo, ii.getAllColumns(), rootFilter);
+ Range<Long> range = TsConditionExtractor.extractTsCondition(ii.getAllColumns().get(tableRecordInfo.getTimestampColumn()), rootFilter);
Assert.assertEquals(946684800000L, range.lowerEndpoint().longValue());
Assert.assertEquals(946771200000L, range.upperEndpoint().longValue());
@@ -133,7 +134,7 @@ public class TsConditionExtractorTest extends LocalFileMetadataTestCase {
LogicalTupleFilter root = new LogicalTupleFilter(TupleFilter.FilterOperatorEnum.AND);
root.addChildren(Lists.newArrayList(subRoot, outFilter));
- Range<Long> range = TsConditionExtractor.extractTsCondition(tableRecordInfo, ii.getAllColumns(), root);
+ Range<Long> range = TsConditionExtractor.extractTsCondition(ii.getAllColumns().get(tableRecordInfo.getTimestampColumn()), root);
Assert.assertEquals(946684800000L, range.lowerEndpoint().longValue());
Assert.assertEquals(946771200000L, range.upperEndpoint().longValue());
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/storage/src/test/java/org/apache/kylin/storage/test/StorageTest.java
----------------------------------------------------------------------
diff --git a/storage/src/test/java/org/apache/kylin/storage/test/StorageTest.java b/storage/src/test/java/org/apache/kylin/storage/test/StorageTest.java
index 0280e85..5212d2e 100644
--- a/storage/src/test/java/org/apache/kylin/storage/test/StorageTest.java
+++ b/storage/src/test/java/org/apache/kylin/storage/test/StorageTest.java
@@ -24,6 +24,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import org.apache.commons.io.IOUtils;
import org.apache.kylin.metadata.realization.SQLDigest;
import org.apache.kylin.storage.IStorageEngine;
@@ -77,7 +78,7 @@ public class StorageTest extends HBaseMetadataTestCase {
CubeManager cubeMgr = CubeManager.getInstance(getTestConfig());
cube = cubeMgr.getCube("TEST_KYLIN_CUBE_WITHOUT_SLR_EMPTY");
Assert.assertNotNull(cube);
- storageEngine = StorageEngineFactory.getStorageEngine(cube);
+ storageEngine = StorageEngineFactory.getStorageEngine(cube,true);
String url = KylinConfig.getInstanceFromEnv().getStorageUrl();
context = new StorageContext();
context.setConnUrl(url);
@@ -162,9 +163,7 @@ public class StorageTest extends HBaseMetadataTestCase {
} catch (Exception e) {
e.printStackTrace();
} finally {
- if (iterator != null) {
- iterator.close();
- }
+ IOUtils.closeQuietly(iterator);
}
return count;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/dea04bf1/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
----------------------------------------------------------------------
diff --git a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java b/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
index 4231848..5ed05e8 100644
--- a/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
+++ b/streaming/src/main/java/org/apache/kylin/streaming/invertedindex/IIStreamBuilder.java
@@ -100,14 +100,14 @@ public class IIStreamBuilder extends StreamBuilder {
return;
}
logger.info("stream build start, size:" + streamsToBuild.size());
- Stopwatch stopwatch = new Stopwatch().start();
+ Stopwatch stopwatch = Stopwatch.createStarted();
final Slice slice = sliceBuilder.buildSlice(streamsToBuild, getStreamParser());
logger.info("slice info, shard:" + slice.getShard() + " timestamp:" + slice.getTimestamp() + " record count:" + slice.getRecordCount());
loadToHBase(hTable, slice, new IIKeyValueCodec(slice.getInfo()));
submitOffset(offset);
stopwatch.stop();
- logger.info("stream build finished, size:" + streamsToBuild.size() + " elapsed time:" + stopwatch.elapsedTime(TimeUnit.MILLISECONDS) + " " + TimeUnit.MILLISECONDS);
+ logger.info("stream build finished, size:" + streamsToBuild.size() + " elapsed time:" + stopwatch.elapsed(TimeUnit.MILLISECONDS) + " " + TimeUnit.MILLISECONDS);
} else {
logger.info("nothing to build, skip building");
}