You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by ma...@apache.org on 2015/08/13 09:05:24 UTC
[04/10] incubator-kylin git commit: KYLIN-909 use kryo to serialized
request
KYLIN-909 use kryo to serialized request
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/4573deeb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/4573deeb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/4573deeb
Branch: refs/heads/KYLIN-909
Commit: 4573deebb26c319c34fa45911fe8aabacc27b55d
Parents: 4e9223b
Author: honma <ho...@ebay.com>
Authored: Wed Aug 5 11:04:27 2015 +0800
Committer: honma <ho...@ebay.com>
Committed: Thu Aug 6 16:05:39 2015 +0800
----------------------------------------------------------------------
core-common/pom.xml | 13 +-
.../org/apache/kylin/common/util/ByteArray.java | 10 +-
.../kylin/common/util/ImmutableBitSet.java | 32 ++++-
.../org/apache/kylin/common/util/KryoUtils.java | 58 ++++++++
.../org/apache/kylin/common/util/BasicTest.java | 40 ++++--
.../apache/kylin/common/util/FatherBean.java | 49 +++++++
.../kylin/cube/gridtable/CubeCodeSystem.java | 67 ++++++++--
.../cube/gridtable/TrimmedCubeCodeSystem.java | 131 +++++++++++++++++++
.../java/org/apache/kylin/gridtable/GTInfo.java | 26 ++--
.../org/apache/kylin/gridtable/GTRecord.java | 70 ++++++++--
.../kylin/gridtable/GTSampleCodeSystem.java | 5 +
.../apache/kylin/gridtable/GTScanRequest.java | 27 ++--
.../apache/kylin/gridtable/IGTCodeSystem.java | 2 +
pom.xml | 6 +
.../kylin/query/test/ITKylinQueryTest.java | 2 +-
.../apache/kylin/storage/StorageFactory.java | 2 +-
.../storage/cube/CubeHBaseEndpointRPC.java | 4 +-
.../apache/kylin/storage/cube/CubeHBaseRPC.java | 9 +-
.../kylin/storage/cube/CubeHBaseScanRPC.java | 27 ++--
.../apache/kylin/storage/cube/CubeScanner.java | 37 ++++--
.../hbase/coprocessor/DictCodeSystem.java | 1 +
21 files changed, 523 insertions(+), 95 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-common/pom.xml
----------------------------------------------------------------------
diff --git a/core-common/pom.xml b/core-common/pom.xml
index 294c57e..6f9f20b 100644
--- a/core-common/pom.xml
+++ b/core-common/pom.xml
@@ -17,7 +17,8 @@
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>kylin-core-common</artifactId>
@@ -34,7 +35,7 @@
</properties>
<dependencies>
- <!-- Basic Utilities -->
+ <!-- Basic Utilities -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
@@ -83,8 +84,12 @@
<groupId>com.ning</groupId>
<artifactId>compress-lzf</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>kryo</artifactId>
+ </dependency>
- <!-- Logging -->
+ <!-- Logging -->
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
@@ -108,6 +113,6 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
-
+
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java b/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
index 460ca69..a388dda 100644
--- a/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
+++ b/core-common/src/main/java/org/apache/kylin/common/util/ByteArray.java
@@ -21,15 +21,13 @@ package org.apache.kylin.common.util;
import java.io.Serializable;
import java.nio.ByteBuffer;
-import org.apache.kylin.common.util.Bytes;
-
/**
* @author yangli9
*/
public class ByteArray implements Comparable<ByteArray>, Serializable {
private static final long serialVersionUID = 1L;
-
+
public static final ByteArray EMPTY = new ImmutableByteArray();
public static ByteArray allocate(int length) {
@@ -160,13 +158,13 @@ public class ByteArray implements Comparable<ByteArray>, Serializable {
else
return Bytes.toStringBinary(data, offset, length);
}
-
+
// ============================================================================
-
+
public static class ImmutableByteArray extends ByteArray {
private static final long serialVersionUID = 1L;
-
+
public ImmutableByteArray() {
super();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-common/src/main/java/org/apache/kylin/common/util/ImmutableBitSet.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/ImmutableBitSet.java b/core-common/src/main/java/org/apache/kylin/common/util/ImmutableBitSet.java
index c401425..b88b279 100644
--- a/core-common/src/main/java/org/apache/kylin/common/util/ImmutableBitSet.java
+++ b/core-common/src/main/java/org/apache/kylin/common/util/ImmutableBitSet.java
@@ -1,5 +1,6 @@
package org.apache.kylin.common.util;
+import java.nio.ByteBuffer;
import java.util.BitSet;
public class ImmutableBitSet {
@@ -22,7 +23,7 @@ public class ImmutableBitSet {
public ImmutableBitSet(int indexFrom, int indexTo) {
this(newBitSet(indexFrom, indexTo));
}
-
+
private static BitSet newBitSet(int indexFrom, int indexTo) {
BitSet set = new BitSet(indexTo);
set.set(indexFrom, indexTo);
@@ -48,7 +49,7 @@ public class ImmutableBitSet {
public int trueBitAt(int i) {
return arr[i];
}
-
+
/** return the bit's index among true bits */
public int trueBitIndexOf(int bitIndex) {
for (int i = 0; i < arr.length; i++) {
@@ -65,7 +66,7 @@ public class ImmutableBitSet {
public ImmutableBitSet set(int bitIndex) {
return set(bitIndex, true);
}
-
+
public ImmutableBitSet set(int bitIndex, boolean value) {
if (set.get(bitIndex) == value) {
return this;
@@ -75,13 +76,13 @@ public class ImmutableBitSet {
return new ImmutableBitSet(mutable);
}
}
-
+
public ImmutableBitSet or(ImmutableBitSet another) {
BitSet mutable = mutable();
mutable.or(another.set);
return new ImmutableBitSet(mutable);
}
-
+
public ImmutableBitSet andNot(ImmutableBitSet another) {
BitSet mutable = mutable();
mutable.andNot(another.set);
@@ -105,7 +106,7 @@ public class ImmutableBitSet {
ImmutableBitSet other = (ImmutableBitSet) obj;
return this.set.equals(other.set);
}
-
+
@Override
public String toString() {
return set.toString();
@@ -128,4 +129,23 @@ public class ImmutableBitSet {
public boolean isEmpty() {
return set.isEmpty();
}
+
+ // SerDer
+ // ============================================================================
+ public static final Serializer SERIALIZER = new Serializer();
+
+ public static class Serializer implements BytesSerializer<ImmutableBitSet> {
+
+ @Override
+ public void serialize(ImmutableBitSet value, ByteBuffer out) {
+ BytesUtil.writeByteArray(value.set.toByteArray(), out);
+ }
+
+ @Override
+ public ImmutableBitSet deserialize(ByteBuffer in) {
+ byte[] bytes = BytesUtil.readByteArray(in);
+ return new ImmutableBitSet(BitSet.valueOf(bytes));
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-common/src/main/java/org/apache/kylin/common/util/KryoUtils.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/util/KryoUtils.java b/core-common/src/main/java/org/apache/kylin/common/util/KryoUtils.java
new file mode 100644
index 0000000..dd2b727
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/util/KryoUtils.java
@@ -0,0 +1,58 @@
+/*
+ * 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.common.util;
+
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public class KryoUtils {
+
+ private static ThreadLocal<Kryo> _Kryo = new ThreadLocal<>();
+
+ public static byte[] serialize(Object obj) {
+ Kryo kryo = getKryo();
+ Output output = new Output(1024, 2014 * 1024);
+ kryo.writeObject(output, obj);
+ return output.toBytes();
+ }
+
+ public static <T> T deserialize(byte[] bytes, Class<T> clazz) {
+ Kryo kryo = getKryo();
+ Input input = new Input(bytes);
+ return kryo.readObject(input, clazz);
+ }
+
+ public static <T> T copy(T origin, Class<T> clazz) {
+ byte[] bytes = serialize(origin);
+ return deserialize(bytes, clazz);
+ }
+
+ private static Kryo getKryo() {
+ if (_Kryo.get() == null) {
+ Kryo kryo = new Kryo();
+ kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy()));
+ _Kryo.set(kryo);
+ }
+
+ return _Kryo.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
index 2f0590b..fe55700 100644
--- a/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
+++ b/core-common/src/test/java/org/apache/kylin/common/util/BasicTest.java
@@ -18,14 +18,6 @@
package org.apache.kylin.common.util;
-import com.google.common.collect.Lists;
-import com.google.common.collect.TreeMultiset;
-import org.apache.commons.configuration.ConfigurationException;
-import org.junit.Assert;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.slf4j.LoggerFactory;
-
import java.io.IOException;
import java.nio.ByteBuffer;
import java.text.DateFormat;
@@ -34,6 +26,19 @@ import java.util.ArrayList;
import java.util.Calendar;
import java.util.IdentityHashMap;
+import org.apache.commons.configuration.ConfigurationException;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+import org.slf4j.LoggerFactory;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import com.google.common.collect.Lists;
+import com.google.common.collect.TreeMultiset;
+
/**
* <p/>
* Keep this test case to test basic java functionality
@@ -63,12 +68,27 @@ public class BasicTest {
@Test
public void test0() throws Exception {
+
+ FatherBean father = new FatherBean();
+ father.age = 50;
+ father.name = "yuan";
+
+ father.init();
+
+ Kryo kryo = new Kryo();
+ kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy()));
+ Output output = new Output(1024, 2014 * 1024);
+ kryo.writeObject(output, father);
+ byte[] persisted = output.toBytes();
+
+ Input input = new Input(persisted);
+ FatherBean newFather = kryo.readObject(input, FatherBean.class);
+
TreeMultiset<Long> xx = TreeMultiset.create();
xx.add(2L);
xx.add(1L);
xx.add(1L);
- for(Long hi : xx)
- {
+ for (Long hi : xx) {
System.out.println(hi);
}
System.out.println(Long.MAX_VALUE);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-common/src/test/java/org/apache/kylin/common/util/FatherBean.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/FatherBean.java b/core-common/src/test/java/org/apache/kylin/common/util/FatherBean.java
new file mode 100644
index 0000000..372a171
--- /dev/null
+++ b/core-common/src/test/java/org/apache/kylin/common/util/FatherBean.java
@@ -0,0 +1,49 @@
+/*
+ * 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.common.util;
+
+import java.util.BitSet;
+
+public class FatherBean {
+ ChildBean child;
+ ChildBean childShadow;
+ String name;
+ BitSet bitset;
+ int age;
+
+ public void init() {
+ child = new ChildBean("bin", 23, this);
+ childShadow = child;
+ bitset = new BitSet(10);
+ bitset.set(4);
+ }
+
+ public class ChildBean {
+ final String name;
+ final int age;
+ final FatherBean father;
+
+ public ChildBean(String name, int age, FatherBean father) {
+ this.name = name;
+ this.age = age;
+ this.father = father;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java
index 54d7525..008bdec 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/CubeCodeSystem.java
@@ -1,5 +1,11 @@
package org.apache.kylin.cube.gridtable;
+import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.ImmutableBitSet;
@@ -15,12 +21,6 @@ import org.apache.kylin.metadata.measure.serializer.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.UnsupportedEncodingException;
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Map;
-
/**
* Created by shaoshi on 3/23/15.
* This implementation uses Dictionary to encode and decode the table; If a column doesn't have dictionary, will check
@@ -32,13 +32,13 @@ public class CubeCodeSystem implements IGTCodeSystem {
// ============================================================================
- private transient GTInfo info;
+ private GTInfo info;
private Map<Integer, Dictionary> dictionaryMap; // column index ==> dictionary of column
private Map<Integer, Integer> fixLenMap; // column index ==> fixed length of column
private Map<Integer, Integer> dependentMetricsMap;
- private IGTComparator comparator;
private DataTypeSerializer[] serializers;
+ private IGTComparator comparator;
public CubeCodeSystem(Map<Integer, Dictionary> dictionaryMap) {
this(dictionaryMap, Collections.<Integer, Integer> emptyMap(), Collections.<Integer, Integer> emptyMap());
@@ -50,6 +50,20 @@ public class CubeCodeSystem implements IGTCodeSystem {
this.dependentMetricsMap = dependentMetricsMap;
}
+ public IGTCodeSystem trim() {
+ DataTypeSerializer[] newSerializers = new DataTypeSerializer[serializers.length];
+
+ for (int i = 0; i < serializers.length; i++) {
+ if (serializers[i] instanceof DictionarySerializer) {
+ newSerializers[i] = new TrimmedDictionarySerialier(serializers[i].maxLength());
+ } else {
+ newSerializers[i] = serializers[i];
+ }
+ }
+
+ return new TrimmedCubeCodeSystem(info, dependentMetricsMap, newSerializers, comparator);
+ }
+
@Override
public void init(GTInfo info) {
this.info = info;
@@ -94,7 +108,7 @@ public class CubeCodeSystem implements IGTCodeSystem {
}
@Override
- public void encodeColumnValue(int col, Object value, int roundingFlag, ByteBuffer buf) {
+ public void encodeColumnValue(int col, Object value, int roundingFlag, ByteBuffer buf) {
DataTypeSerializer serializer = serializers[col];
if (serializer instanceof DictionarySerializer) {
((DictionarySerializer) serializer).serializeWithRounding(value, roundingFlag, buf);
@@ -139,6 +153,41 @@ public class CubeCodeSystem implements IGTCodeSystem {
return result;
}
+ static class TrimmedDictionarySerialier extends DataTypeSerializer {
+
+ final int fieldSize;
+
+ public TrimmedDictionarySerialier(int fieldSize) {
+ this.fieldSize = fieldSize;
+
+ }
+
+ @Override
+ public int peekLength(ByteBuffer in) {
+ return fieldSize;
+ }
+
+ @Override
+ public int maxLength() {
+ return fieldSize;
+ }
+
+ @Override
+ public Object valueOf(byte[] value) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void serialize(Object value, ByteBuffer out) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object deserialize(ByteBuffer in) {
+ throw new UnsupportedOperationException();
+ }
+ }
+
static class DictionarySerializer extends DataTypeSerializer {
private Dictionary dictionary;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java
new file mode 100644
index 0000000..e603eaf
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/cube/gridtable/TrimmedCubeCodeSystem.java
@@ -0,0 +1,131 @@
+/*
+ *
+ * * 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.cube.gridtable;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.dict.Dictionary;
+import org.apache.kylin.dict.DictionarySerializer;
+import org.apache.kylin.gridtable.DefaultGTComparator;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.IGTCodeSystem;
+import org.apache.kylin.gridtable.IGTComparator;
+import org.apache.kylin.metadata.measure.MeasureAggregator;
+import org.apache.kylin.metadata.measure.serializer.DataTypeSerializer;
+import org.apache.kylin.metadata.measure.serializer.StringSerializer;
+
+@SuppressWarnings({ "rawtypes", "unchecked" })
+public class TrimmedCubeCodeSystem implements IGTCodeSystem {
+
+ private GTInfo info;
+
+ private Map<Integer, Integer> dependentMetricsMap;
+ private DataTypeSerializer[] serializers;
+ private IGTComparator comparator;
+
+ public TrimmedCubeCodeSystem(GTInfo info, Map<Integer, Integer> dependentMetricsMap, DataTypeSerializer[] serializers, IGTComparator comparator) {
+ this.info = info;
+ this.dependentMetricsMap = dependentMetricsMap;
+ this.serializers = serializers;
+ this.comparator = comparator;
+ }
+
+ @Override
+ public void init(GTInfo info) {
+ }
+
+ @Override
+ public IGTCodeSystem trim() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public IGTComparator getComparator() {
+ return comparator;
+ }
+
+ @Override
+ public int codeLength(int col, ByteBuffer buf) {
+ return serializers[col].peekLength(buf);
+ }
+
+ @Override
+ public int maxCodeLength(int col) {
+ return serializers[col].maxLength();
+ }
+
+ @Override
+ public void encodeColumnValue(int col, Object value, ByteBuffer buf) {
+ encodeColumnValue(col, value, 0, buf);
+ }
+
+ @Override
+ public void encodeColumnValue(int col, Object value, int roundingFlag, ByteBuffer buf) {
+ DataTypeSerializer serializer = serializers[col];
+ if (serializer instanceof CubeCodeSystem.TrimmedDictionarySerialier || serializer instanceof CubeCodeSystem.DictionarySerializer) {
+ //TODO: remove this check
+ throw new IllegalStateException("Encode dictionary value in coprocessor");
+ } else {
+ if ((!(serializer instanceof StringSerializer || serializer instanceof CubeCodeSystem.FixLenSerializer)) && (value instanceof String)) {
+ value = serializer.valueOf((String) value);
+ }
+ serializer.serialize(value, buf);
+ }
+ }
+
+ @Override
+ public Object decodeColumnValue(int col, ByteBuffer buf) {
+ return serializers[col].deserialize(buf);
+ }
+
+ //TODO: remove duplicate
+ @Override
+ public MeasureAggregator<?>[] newMetricsAggregators(ImmutableBitSet columns, String[] aggrFunctions) {
+ assert columns.trueBitCount() == aggrFunctions.length;
+
+ MeasureAggregator<?>[] result = new MeasureAggregator[aggrFunctions.length];
+ for (int i = 0; i < result.length; i++) {
+ int col = columns.trueBitAt(i);
+ result[i] = MeasureAggregator.create(aggrFunctions[i], info.getColumnType(col).toString());
+ }
+
+ // deal with holistic distinct count
+ if (dependentMetricsMap != null) {
+ for (Integer child : dependentMetricsMap.keySet()) {
+ if (columns.get(child)) {
+ Integer parent = dependentMetricsMap.get(child);
+ if (columns.get(parent) == false)
+ throw new IllegalStateException();
+
+ int childIdx = columns.trueBitIndexOf(child);
+ int parentIdx = columns.trueBitIndexOf(parent);
+ result[childIdx].setDependentAggregator(result[parentIdx]);
+ }
+ }
+ }
+
+ return result;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java
index ff748ec..e0cf34c 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTInfo.java
@@ -1,14 +1,17 @@
package org.apache.kylin.gridtable;
-import org.apache.kylin.common.util.ImmutableBitSet;
-import org.apache.kylin.metadata.model.DataType;
-import org.apache.kylin.metadata.model.TblColRef;
-
import java.util.Arrays;
import java.util.BitSet;
import java.util.Iterator;
import java.util.LinkedList;
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.common.util.KryoUtils;
+import org.apache.kylin.metadata.model.DataType;
+import org.apache.kylin.metadata.model.TblColRef;
+
+import com.google.common.base.Preconditions;
+
public class GTInfo {
public static Builder builder() {
@@ -16,7 +19,7 @@ public class GTInfo {
}
String tableName;
-
+
IGTCodeSystem codeSystem;
// column schema
@@ -39,12 +42,17 @@ public class GTInfo {
private GTInfo() {
}
- public String getTableName() {
- return tableName;
+ public byte[] serializedTrimmedVersion() {
+ IGTCodeSystem originalCodeSystem = this.codeSystem;
+ codeSystem = this.codeSystem.trim();
+ byte[] bytes = KryoUtils.serialize(this);
+ this.codeSystem = originalCodeSystem;
+ return bytes;
}
- public IGTCodeSystem getCodeSystem() {
- return codeSystem;
+
+ public String getTableName() {
+ return tableName;
}
public int getColumnCount() {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
index ef2efe0..6a1a5e7 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTRecord.java
@@ -5,6 +5,8 @@ import java.util.Arrays;
import java.util.BitSet;
import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.BytesSerializer;
+import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.ImmutableBitSet;
public class GTRecord implements Comparable<GTRecord> {
@@ -14,15 +16,22 @@ public class GTRecord implements Comparable<GTRecord> {
private ImmutableBitSet maskForEqualHashComp;
- public GTRecord(GTInfo info, ImmutableBitSet maskForEqualHashComp) {
+ public GTRecord(GTInfo info, ImmutableBitSet maskForEqualHashComp, ByteArray[] cols) {
this.info = info;
+ this.cols = cols;
+ this.maskForEqualHashComp = maskForEqualHashComp;
+ }
+
+ public GTRecord(GTInfo info, ImmutableBitSet maskForEqualHashComp) {
this.cols = new ByteArray[info.getColumnCount()];
- for (int i = 0; i < cols.length; i++) {
+ for (int i = 0; i < this.cols.length; i++) {
if (maskForEqualHashComp.get(i)) {
this.cols[i] = new ByteArray();
}
}
+ this.info = info;
this.maskForEqualHashComp = maskForEqualHashComp;
+
}
public GTRecord(GTInfo info) {
@@ -33,6 +42,7 @@ public class GTRecord implements Comparable<GTRecord> {
return info;
}
+
public ByteArray get(int i) {
return cols[i];
}
@@ -50,7 +60,7 @@ public class GTRecord implements Comparable<GTRecord> {
/** set record to the codes of specified values, reuse given space to hold the codes */
public GTRecord setValues(ImmutableBitSet selectedCols, ByteArray space, Object... values) {
assert selectedCols.cardinality() == values.length;
-
+
ByteBuffer buf = space.asBuffer();
int pos = buf.position();
for (int i = 0; i < selectedCols.trueBitCount(); i++) {
@@ -71,7 +81,7 @@ public class GTRecord implements Comparable<GTRecord> {
/** decode and return the values of this record */
public Object[] getValues(ImmutableBitSet selectedCols, Object[] result) {
assert selectedCols.cardinality() == result.length;
-
+
for (int i = 0; i < selectedCols.trueBitCount(); i++) {
int c = selectedCols.trueBitAt(i);
if (cols[c] == null || cols[c].array() == null) {
@@ -182,7 +192,7 @@ public class GTRecord implements Comparable<GTRecord> {
public String toString() {
return toString(maskForEqualHashComp);
}
-
+
public String toString(ImmutableBitSet selectedColumns) {
Object[] values = new Object[selectedColumns.cardinality()];
getValues(selectedColumns, values);
@@ -228,7 +238,6 @@ public class GTRecord implements Comparable<GTRecord> {
}
}
-
/** write data to given buffer, like serialize */
public void exportColumnBlock(int c, ByteBuffer buf) {
exportColumns(info.colBlocks[c], buf);
@@ -260,9 +269,9 @@ public class GTRecord implements Comparable<GTRecord> {
public static ByteArray exportScanKey(GTRecord rec) {
if (rec == null)
return null;
-
+
GTInfo info = rec.getInfo();
-
+
BitSet selectedColumns = new BitSet();
int len = 0;
for (int i = 0; i < info.primaryKey.trueBitCount(); i++) {
@@ -273,7 +282,7 @@ public class GTRecord implements Comparable<GTRecord> {
selectedColumns.set(c);
len += rec.cols[c].length();
}
-
+
if (selectedColumns.cardinality() == 0)
return null;
@@ -281,5 +290,46 @@ public class GTRecord implements Comparable<GTRecord> {
rec.exportColumns(new ImmutableBitSet(selectedColumns), buf);
return buf;
}
-
+
+ // SerDer
+ // ============================================================================
+
+ public static final Serializer serializer = new Serializer();
+
+ //will ignore GTInfo
+ private static class Serializer implements BytesSerializer<GTRecord> {
+
+ @Override
+ public void serialize(GTRecord value, ByteBuffer out) {
+ //value.clone()
+ BytesUtil.writeVInt(value.cols.length, out);
+ for (int i = 0; i < value.cols.length; i++) {
+ if (value.cols[i] == null) {
+ BytesUtil.writeVInt(0, out);
+ } else {
+ BytesUtil.writeVInt(1, out);
+ BytesUtil.writeByteArray(value.cols[i].array(), value.cols[i].offset(), value.cols[i].length(), out);
+ }
+ }
+ ImmutableBitSet.SERIALIZER.serialize(value.maskForEqualHashComp, out);
+ }
+
+ @Override
+ public GTRecord deserialize(ByteBuffer in) {
+ int colLength = BytesUtil.readVInt(in);
+ ByteArray[] cols = new ByteArray[colLength];
+ for (int i = 0; i < colLength; i++) {
+ int flag = BytesUtil.readVInt(in);
+ if (flag == 1) {
+ byte[] bytes = BytesUtil.readByteArray(in);
+ cols[i] = new ByteArray(bytes, 0, bytes.length);
+ } else {
+ cols[i] = null;
+ }
+ }
+ ImmutableBitSet maskForEqualHashComp = ImmutableBitSet.SERIALIZER.deserialize(in);
+ return new GTRecord(null, maskForEqualHashComp, cols);
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java
index 83b0ed8..08b4b61 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTSampleCodeSystem.java
@@ -37,6 +37,11 @@ public class GTSampleCodeSystem implements IGTCodeSystem {
}
@Override
+ public IGTCodeSystem trim() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
public int codeLength(int col, ByteBuffer buf) {
return serializers[col].peekLength(buf);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java
index 30ece92..749ebde 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTScanRequest.java
@@ -1,18 +1,17 @@
package org.apache.kylin.gridtable;
-import com.google.common.collect.Sets;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Set;
+
import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.model.TblColRef;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Set;
+import com.google.common.collect.Sets;
public class GTScanRequest {
- // basic
- private GTInfo info;
private GTScanRange range;
private ImmutableBitSet columns;
private ImmutableBitSet selectedColBlocks;
@@ -33,11 +32,10 @@ public class GTScanRequest {
}
public GTScanRequest(GTInfo info, GTScanRange range, ImmutableBitSet columns, TupleFilter filterPushDown) {
- this.info = info;
this.range = range == null ? new GTScanRange(new GTRecord(info), new GTRecord(info)) : range;
this.columns = columns;
this.filterPushDown = filterPushDown;
- validate();
+ validate(info);
}
public GTScanRequest(GTInfo info, GTScanRange range, ImmutableBitSet aggrGroupBy, ImmutableBitSet aggrMetrics, //
@@ -47,7 +45,6 @@ public class GTScanRequest {
public GTScanRequest(GTInfo info, GTScanRange range, ImmutableBitSet dimensions, ImmutableBitSet aggrGroupBy, //
ImmutableBitSet aggrMetrics, String[] aggrMetricsFuncs, TupleFilter filterPushDown, boolean allowPreAggregation) {
- this.info = info;
this.range = range == null ? new GTScanRange(new GTRecord(info), new GTRecord(info)) : range;
this.columns = dimensions;
this.filterPushDown = filterPushDown;
@@ -58,10 +55,10 @@ public class GTScanRequest {
this.allowPreAggregation = allowPreAggregation;
- validate();
+ validate(info);
}
- private void validate() {
+ private void validate(GTInfo info) {
if (range == null)
range = new GTScanRange(null, null);
@@ -83,11 +80,11 @@ public class GTScanRequest {
this.selectedColBlocks = info.selectColumnBlocks(columns);
if (hasFilterPushDown()) {
- validateFilterPushDown();
+ validateFilterPushDown(info);
}
}
- private void validateFilterPushDown() {
+ private void validateFilterPushDown(GTInfo info) {
if (hasFilterPushDown() == false)
return;
@@ -135,10 +132,6 @@ public class GTScanRequest {
return aggrGroupBy != null && aggrMetrics != null && aggrMetricsFuncs != null;
}
- public GTInfo getInfo() {
- return info;
- }
-
public GTRecord getPkStart() {
return range.pkStart;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java
index bb715b2..c4c3692 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTCodeSystem.java
@@ -9,6 +9,8 @@ public interface IGTCodeSystem {
void init(GTInfo info);
+ IGTCodeSystem trim();
+
IGTComparator getComparator();
/** Return the length of code starting at the specified buffer, buffer position must not change after return */
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 2bf345a..38c39bc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -86,6 +86,7 @@
<compress-lzf.version>1.0.3</compress-lzf.version>
<extendedset.version>1.3.4</extendedset.version>
<jetty.version>9.2.7.v20150116</jetty.version>
+ <kryo.version>3.0.3</kryo.version>
<!-- REST Service -->
<spring.framework.version>3.1.2.RELEASE</spring.framework.version>
@@ -419,6 +420,11 @@
<artifactId>extendedset</artifactId>
<version>${extendedset.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>kryo</artifactId>
+ <version>${kryo.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.curator</groupId>
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/query/src/test/java/org/apache/kylin/query/test/ITKylinQueryTest.java
----------------------------------------------------------------------
diff --git a/query/src/test/java/org/apache/kylin/query/test/ITKylinQueryTest.java b/query/src/test/java/org/apache/kylin/query/test/ITKylinQueryTest.java
index 6d192e9..1ba2125 100644
--- a/query/src/test/java/org/apache/kylin/query/test/ITKylinQueryTest.java
+++ b/query/src/test/java/org/apache/kylin/query/test/ITKylinQueryTest.java
@@ -95,7 +95,7 @@ public class ITKylinQueryTest extends KylinTestBase {
@Test
public void testSingleRunQuery() throws Exception {
- String queryFileName = "src/test/resources/query/sql/query58.sql";
+ String queryFileName = "src/test/resources/query/sql/query44.sql";
File sqlFile = new File(queryFileName);
if (sqlFile.exists()) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java b/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
index f3dd900..77e3010 100644
--- a/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
+++ b/storage/src/main/java/org/apache/kylin/storage/StorageFactory.java
@@ -29,7 +29,7 @@ import org.apache.kylin.metadata.realization.IRealization;
import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.storage.cache.CacheFledgedDynamicStorageEngine;
import org.apache.kylin.storage.cache.CacheFledgedStaticStorageEngine;
-import org.apache.kylin.storage.cube.CubeStorageQuery;
+import org.apache.kylin.storage.hbase.CubeStorageQuery;
import org.apache.kylin.storage.hbase.InvertedIndexStorageQuery;
import org.apache.kylin.storage.hybrid.HybridInstance;
import org.apache.kylin.storage.hybrid.HybridStorageEngine;
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java
index 8c7b24c..ec2d223 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseEndpointRPC.java
@@ -59,7 +59,7 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
final ResultScanner scanner = hbaseTable.getScanner(hbaseScan);
final Iterator<Result> iterator = scanner.iterator();
- final GTRowBlock oneBlock = new GTRowBlock(scanRequest.getInfo()); // avoid object creation
+ final GTRowBlock oneBlock = new GTRowBlock(info); // avoid object creation
oneBlock.setNumberOfRows(1); // row block is always disabled for cubes, row block contains only one record
IGTStore store = new IGTStore() {
@@ -120,7 +120,7 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
}
};
- GTRawScanner rawScanner = new GTRawScanner(scanRequest.getInfo(), store, scanRequest);
+ GTRawScanner rawScanner = new GTRawScanner(info, store, scanRequest);
return scanRequest.decorateScanner(rawScanner);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
index 0045355..0f51813 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseRPC.java
@@ -1,6 +1,9 @@
package org.apache.kylin.storage.cube;
-import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
import org.apache.hadoop.hbase.client.Scan;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
@@ -19,9 +22,7 @@ import org.apache.kylin.gridtable.IGTScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
+import com.google.common.collect.Lists;
public abstract class CubeHBaseRPC {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
index e4035ac..b30dd0c 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeHBaseScanRPC.java
@@ -1,19 +1,29 @@
package org.apache.kylin.storage.cube;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.client.HConnection;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.common.util.KryoUtils;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.gridtable.*;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRawScanner;
+import org.apache.kylin.gridtable.GTRowBlock;
+import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.gridtable.IGTScanner;
+import org.apache.kylin.gridtable.IGTStore;
import org.apache.kylin.storage.hbase.HBaseConnection;
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.List;
-
/**
* for test use only
*/
@@ -21,6 +31,7 @@ public class CubeHBaseScanRPC extends CubeHBaseRPC {
public CubeHBaseScanRPC(GTInfo info, CubeSegment cubeSeg, Cuboid cuboid) {
super(info, cubeSeg, cuboid);
+
}
@Override
@@ -39,7 +50,7 @@ public class CubeHBaseScanRPC extends CubeHBaseRPC {
final ResultScanner scanner = hbaseTable.getScanner(hbaseScan);
final Iterator<Result> iterator = scanner.iterator();
- final GTRowBlock oneBlock = new GTRowBlock(scanRequest.getInfo()); // avoid object creation
+ final GTRowBlock oneBlock = new GTRowBlock(info); // avoid object creation
oneBlock.setNumberOfRows(1); // row block is always disabled for cubes, row block contains only one record
IGTStore store = new IGTStore() {
@@ -100,7 +111,7 @@ public class CubeHBaseScanRPC extends CubeHBaseRPC {
}
};
- GTRawScanner rawScanner = new GTRawScanner(scanRequest.getInfo(), store, scanRequest);
+ GTRawScanner rawScanner = new GTRawScanner(info, store, scanRequest);
//return rawScanner;
return scanRequest.decorateScanner(rawScanner);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java b/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
index d930d87..b93e522 100644
--- a/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/CubeScanner.java
@@ -1,20 +1,35 @@
package org.apache.kylin.storage.cube;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
+import java.io.IOException;
+import java.util.BitSet;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+
import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.common.util.KryoUtils;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.gridtable.CubeGridTable;
import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
import org.apache.kylin.cube.model.CubeDesc;
-import org.apache.kylin.gridtable.*;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRecord;
+import org.apache.kylin.gridtable.GTScanRange;
+import org.apache.kylin.gridtable.GTScanRangePlanner;
+import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.gridtable.GTUtil;
+import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
-import java.io.IOException;
-import java.util.*;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
public class CubeScanner implements IGTScanner {
@@ -22,17 +37,20 @@ public class CubeScanner implements IGTScanner {
final CubeSegment cubeSeg;
final GTInfo info;
+ final byte[] trimmedInfo;
final List<GTScanRequest> scanRequests;
final Scanner scanner;
final Cuboid cuboid;
public CubeScanner(CubeSegment cubeSeg, Cuboid cuboid, Set<TblColRef> dimensions, Set<TblColRef> groups, //
- Collection<FunctionDesc> metrics, TupleFilter filter,boolean allowPreAggregate) {
+ Collection<FunctionDesc> metrics, TupleFilter filter, boolean allowPreAggregate) {
this.cuboid = cuboid;
this.cubeSeg = cubeSeg;
this.info = CubeGridTable.newGTInfo(cubeSeg, cuboid.getId());
CuboidToGridTableMapping mapping = cuboid.getCuboidToGridTableMapping();
+
+ //replace the constant values in filter to dictionary codes
TupleFilter gtFilter = GTUtil.convertFilterColumnsAndConstants(filter, info, mapping.getCuboidDimensionsInGTOrder(), groups);
ImmutableBitSet gtDimensions = makeGridTableColumns(mapping, dimensions);
@@ -40,14 +58,16 @@ public class CubeScanner implements IGTScanner {
ImmutableBitSet gtAggrMetrics = makeGridTableColumns(mapping, metrics);
String[] gtAggrFuncs = makeAggrFuncs(mapping, metrics);
+ //TODO: should remove this in endpoint scenario
GTScanRangePlanner scanRangePlanner = new GTScanRangePlanner(info);
List<GTScanRange> scanRanges = scanRangePlanner.planScanRanges(gtFilter, MAX_SCAN_RANGES);
scanRequests = Lists.newArrayListWithCapacity(scanRanges.size());
for (GTScanRange range : scanRanges) {
- scanRequests.add(new GTScanRequest(info, range, gtDimensions, gtAggrGroups, gtAggrMetrics, gtAggrFuncs, gtFilter,allowPreAggregate));
+ scanRequests.add(new GTScanRequest(info, range, gtDimensions, gtAggrGroups, gtAggrMetrics, gtAggrFuncs, gtFilter, allowPreAggregate));
}
+ trimmedInfo = info.serializedTrimmedVersion();
scanner = new Scanner();
}
@@ -152,7 +172,8 @@ public class CubeScanner implements IGTScanner {
return false;
try {
- CubeHBaseScanRPC rpc = new CubeHBaseScanRPC(info, cubeSeg, cuboid);
+ GTInfo x = KryoUtils.deserialize(trimmedInfo,GTInfo.class);
+ CubeHBaseScanRPC rpc = new CubeHBaseScanRPC(x, cubeSeg, cuboid);
inputScanners[cur] = rpc.getGTScanner(scanRequests.get(cur));
curIterator = inputScanners[cur].iterator();
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4573deeb/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
index 563f346..f4fd65d 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/DictCodeSystem.java
@@ -37,6 +37,7 @@ public class DictCodeSystem implements IFilterCodeSystem<String> {
return tupleValue.compareTo(constValue);
}
+ //TODO: should use ISO-8859-1 rather than UTF8
@Override
public void serialize(String value, ByteBuffer buffer) {
BytesUtil.writeUTFString(value, buffer);