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/08/19 13:25:39 UTC
incubator-kylin git commit: KYLIN-951 Refactor IGTStore,
replace RowBlock writer with GTRecord writer
Repository: incubator-kylin
Updated Branches:
refs/heads/0.8 70d1b9d86 -> 4741004cb
KYLIN-951 Refactor IGTStore, replace RowBlock writer with GTRecord writer
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/4741004c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/4741004c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/4741004c
Branch: refs/heads/0.8
Commit: 4741004cb3c3504126e43564f3bc8d7338109ac1
Parents: 70d1b9d
Author: Yang Li <li...@apache.org>
Authored: Wed Aug 19 19:23:58 2015 +0800
Committer: Yang Li <li...@apache.org>
Committed: Wed Aug 19 19:23:58 2015 +0800
----------------------------------------------------------------------
core-cube/pom.xml | 4 +
.../kylin/cube/gridtable/CubeCodeSystem.java | 2 +-
.../cube/gridtable/TrimmedCubeCodeSystem.java | 5 -
.../cube/inmemcubing/ConcurrentDiskStore.java | 125 ++++++++++-------
.../cube/inmemcubing/InMemCubeBuilder.java | 5 -
.../kylin/cube/inmemcubing/MemDiskStore.java | 124 ++++++++++-------
.../kylin/cube/util/BitSetSerializer.java | 62 +++++++++
.../org/apache/kylin/cube/util/KryoUtils.java | 61 ++++++++
.../kylin/gridtable/GTAggregateScanner.java | 5 -
.../org/apache/kylin/gridtable/GTBuilder.java | 40 +-----
.../apache/kylin/gridtable/GTFilterScanner.java | 5 -
.../java/org/apache/kylin/gridtable/GTInfo.java | 32 +++--
.../org/apache/kylin/gridtable/GTRecord.java | 7 +-
.../kylin/gridtable/GTSampleCodeSystem.java | 5 -
.../kylin/gridtable/GTStoreBridgeScanner.java | 111 ---------------
.../apache/kylin/gridtable/IGTCodeSystem.java | 2 -
.../org/apache/kylin/gridtable/IGTScanner.java | 2 -
.../org/apache/kylin/gridtable/IGTStore.java | 13 +-
.../org/apache/kylin/gridtable/IGTWriter.java | 9 ++
.../gridtable/memstore/GTSimpleMemStore.java | 105 +++++++-------
.../kylin/gridtable/SimpleGridTableTest.java | 11 --
storage/pom.xml | 4 -
.../storage/cube/CubeHBaseEndpointRPC.java | 10 +-
.../kylin/storage/cube/CubeHBaseScanRPC.java | 2 +-
.../apache/kylin/storage/cube/CubeScanner.java | 24 +---
.../apache/kylin/storage/cube/HBaseGTStore.java | 119 ----------------
.../kylin/storage/cube/HBaseReadonlyStore.java | 138 +++++++++++++++++++
.../coprocessor/endpoint/CubeVisitService.java | 6 +-
.../kylin/storage/util/BitSetSerializer.java | 62 ---------
.../apache/kylin/storage/util/KryoUtils.java | 61 --------
30 files changed, 519 insertions(+), 642 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/pom.xml
----------------------------------------------------------------------
diff --git a/core-cube/pom.xml b/core-cube/pom.xml
index cf40bf0..68f74a1 100644
--- a/core-cube/pom.xml
+++ b/core-cube/pom.xml
@@ -48,6 +48,10 @@
<groupId>com.n3twork.druid</groupId>
<artifactId>extendedset</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>kryo-shaded</artifactId>
+ </dependency>
<!-- Env & Test -->
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 cdad8b7..e52a6e1 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
@@ -50,7 +50,7 @@ public class CubeCodeSystem implements IGTCodeSystem {
this.dependentMetricsMap = dependentMetricsMap;
}
- public IGTCodeSystem trim() {
+ public IGTCodeSystem trimForCoprocessor() {
DataTypeSerializer[] newSerializers = new DataTypeSerializer[serializers.length];
for (int i = 0; i < serializers.length; i++) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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
index dae44cc..e662a82 100644
--- 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
@@ -52,11 +52,6 @@ public class TrimmedCubeCodeSystem implements IGTCodeSystem {
}
@Override
- public IGTCodeSystem trim() {
- throw new UnsupportedOperationException();
- }
-
- @Override
public IGTComparator getComparator() {
return comparator;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConcurrentDiskStore.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConcurrentDiskStore.java b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConcurrentDiskStore.java
index 5c1a7ee..ebff9c8 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConcurrentDiskStore.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/ConcurrentDiskStore.java
@@ -30,15 +30,16 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.NoSuchElementException;
import org.apache.commons.io.IOUtils;
import org.apache.kylin.gridtable.GTInfo;
-import org.apache.kylin.gridtable.GTRowBlock;
+import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRequest;
-import org.apache.kylin.gridtable.GTStoreBridgeScanner;
import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.gridtable.IGTStore;
+import org.apache.kylin.gridtable.IGTWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -91,17 +92,16 @@ public class ConcurrentDiskStore implements IGTStore, Closeable {
}
@Override
- public IGTStoreWriter rebuild(int shard) throws IOException {
+ public IGTWriter rebuild(int shard) throws IOException {
return newWriter(0);
}
@Override
- public IGTStoreWriter append(int shard, GTRowBlock.Writer fillLast) throws IOException {
- throw new IllegalStateException("does not support append yet");
- //return newWriter(diskFile.length());
+ public IGTWriter append(int shard) throws IOException {
+ return newWriter(diskFile.length());
}
- private IGTStoreWriter newWriter(long startOffset) throws IOException {
+ private IGTWriter newWriter(long startOffset) throws IOException {
synchronized (lock) {
if (activeWriter != null || !activeReaders.isEmpty())
throw new IllegalStateException();
@@ -124,14 +124,10 @@ public class ConcurrentDiskStore implements IGTStore, Closeable {
@Override
public IGTScanner scan(GTScanRequest scanRequest) throws IOException {
- return new GTStoreBridgeScanner(info, scanRequest, oldScan(scanRequest));
- }
-
- private IGTStoreScanner oldScan(GTScanRequest scanRequest) throws IOException {
return newReader();
}
- private IGTStoreScanner newReader() throws IOException {
+ private IGTScanner newReader() throws IOException {
synchronized (lock) {
if (activeWriter != null)
throw new IllegalStateException();
@@ -154,13 +150,11 @@ public class ConcurrentDiskStore implements IGTStore, Closeable {
}
}
- private class Reader implements IGTStoreScanner {
+ private class Reader implements IGTScanner {
final DataInputStream din;
long fileLen;
long readOffset;
-
- GTRowBlock block = GTRowBlock.allocate(info);
- GTRowBlock next = null;
+ int count;
Reader(long startOffset) throws IOException {
this.fileLen = diskFile.length();
@@ -208,55 +202,83 @@ public class ConcurrentDiskStore implements IGTStore, Closeable {
}
@Override
- public boolean hasNext() {
- if (next != null)
- return true;
-
- try {
- if (din.available() > 0) {
- block.importFrom(din);
- next = block;
- }
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
+ public void close() throws IOException {
+ din.close();
+ closeReader(this);
- return next != null;
+ if (debug)
+ logger.debug(ConcurrentDiskStore.this + " read end @ " + readOffset);
}
@Override
- public GTRowBlock next() {
- if (next == null) {
- hasNext();
- if (next == null)
- throw new NoSuchElementException();
- }
- GTRowBlock r = next;
- next = null;
- return r;
+ public Iterator<GTRecord> iterator() {
+ count = 0;
+ return new Iterator<GTRecord>() {
+ GTRecord record = new GTRecord(info);
+ GTRecord next;
+ ByteBuffer buf = ByteBuffer.allocate(info.getMaxRecordLength());
+
+ @Override
+ public boolean hasNext() {
+ if (next != null)
+ return true;
+
+ try {
+ if (din.available() > 0) {
+ int len = din.readInt();
+ din.read(buf.array(), buf.arrayOffset(), len);
+ buf.clear();
+ buf.limit(len);
+ record.loadColumns(info.getAllColumns(), buf);
+ next = record;
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ return next != null;
+ }
+
+ @Override
+ public GTRecord next() {
+ if (next == null) {
+ hasNext();
+ if (next == null)
+ throw new NoSuchElementException();
+ }
+ GTRecord r = next;
+ next = null;
+ count++;
+ return r;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
}
@Override
- public void remove() {
- throw new UnsupportedOperationException();
+ public GTInfo getInfo() {
+ return info;
}
@Override
- public void close() throws IOException {
- din.close();
- closeReader(this);
-
- if (debug)
- logger.debug(ConcurrentDiskStore.this + " read end @ " + readOffset);
+ public int getScannedRowCount() {
+ return count;
}
+
}
- private class Writer implements IGTStoreWriter {
+ private class Writer implements IGTWriter {
final DataOutputStream dout;
+ final ByteBuffer buf;
long writeOffset;
Writer(long startOffset) {
this.writeOffset = startOffset;
+ this.buf = ByteBuffer.allocate(info.getMaxRecordLength());
if (debug)
logger.debug(ConcurrentDiskStore.this + " write start @ " + writeOffset);
@@ -284,8 +306,13 @@ public class ConcurrentDiskStore implements IGTStore, Closeable {
}
@Override
- public void write(GTRowBlock block) throws IOException {
- block.export(dout);
+ public void write(GTRecord rec) throws IOException {
+ buf.clear();
+ rec.exportColumns(info.getAllColumns(), buf);
+
+ int len = buf.position();
+ dout.writeInt(len);
+ dout.write(buf.array(), buf.arrayOffset(), len);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/InMemCubeBuilder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/InMemCubeBuilder.java b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/InMemCubeBuilder.java
index 6e7725d..465dd99 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/InMemCubeBuilder.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/InMemCubeBuilder.java
@@ -580,11 +580,6 @@ public class InMemCubeBuilder extends AbstractInMemCubeBuilder {
return 0;
}
- @Override
- public int getScannedRowBlockCount() {
- return 0;
- }
-
private void buildGTRecord(List<String> row, GTRecord record) {
Object[] dimensions = buildKey(row);
Object[] metricsValues = buildValue(row);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/MemDiskStore.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/MemDiskStore.java b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/MemDiskStore.java
index d76421f..2a12d1b 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/MemDiskStore.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/inmemcubing/MemDiskStore.java
@@ -17,7 +17,7 @@
package org.apache.kylin.cube.inmemcubing;
-import static org.apache.kylin.common.util.MemoryBudgetController.ONE_MB;
+import static org.apache.kylin.common.util.MemoryBudgetController.*;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
@@ -31,17 +31,18 @@ import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
+import java.util.Iterator;
import java.util.NoSuchElementException;
import org.apache.kylin.common.util.MemoryBudgetController;
import org.apache.kylin.common.util.MemoryBudgetController.MemoryConsumer;
import org.apache.kylin.common.util.MemoryBudgetController.NotEnoughBudgetException;
import org.apache.kylin.gridtable.GTInfo;
-import org.apache.kylin.gridtable.GTRowBlock;
+import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRequest;
-import org.apache.kylin.gridtable.GTStoreBridgeScanner;
import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.gridtable.IGTStore;
+import org.apache.kylin.gridtable.IGTWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -87,12 +88,12 @@ public class MemDiskStore implements IGTStore, Closeable {
}
@Override
- public IGTStoreWriter rebuild(int shard) throws IOException {
+ public IGTWriter rebuild(int shard) throws IOException {
return newWriter(0);
}
@Override
- public IGTStoreWriter append(int shard, GTRowBlock.Writer fillLast) throws IOException {
+ public IGTWriter append(int shard) throws IOException {
return newWriter(length());
}
@@ -108,10 +109,6 @@ public class MemDiskStore implements IGTStore, Closeable {
@Override
public IGTScanner scan(GTScanRequest scanRequest) throws IOException {
- return new GTStoreBridgeScanner(info, scanRequest, oldScan(scanRequest));
- }
-
- public IGTStoreScanner oldScan(GTScanRequest scanRequest) throws IOException {
synchronized (lock) {
return new Reader();
}
@@ -135,16 +132,14 @@ public class MemDiskStore implements IGTStore, Closeable {
return "MemDiskStore@" + (info.getTableName() == null ? this.hashCode() : info.getTableName());
}
- private class Reader implements IGTStoreScanner {
+ private class Reader implements IGTScanner {
final DataInputStream din;
long readOffset = 0;
long memRead = 0;
long diskRead = 0;
int nReadCalls = 0;
-
- GTRowBlock block = GTRowBlock.allocate(info);
- GTRowBlock next = null;
+ int count = 0;
Reader() throws IOException {
diskPart.openRead();
@@ -218,54 +213,81 @@ public class MemDiskStore implements IGTStore, Closeable {
}
@Override
- public boolean hasNext() {
- if (next != null)
- return true;
-
- try {
- if (din.available() > 0) {
- block.importFrom(din);
- next = block;
- }
- } catch (IOException e) {
- throw new RuntimeException(e);
+ public void close() throws IOException {
+ synchronized (lock) {
+ din.close();
+ diskPart.closeRead();
+ if (debug)
+ logger.debug(MemDiskStore.this + " read end @ " + readOffset + ", " + (memRead) + " from mem, " + (diskRead) + " from disk, " + nReadCalls + " read() calls");
}
-
- return next != null;
}
@Override
- public GTRowBlock next() {
- if (next == null) {
- hasNext();
- if (next == null)
- throw new NoSuchElementException();
- }
- GTRowBlock r = next;
- next = null;
- return r;
+ public Iterator<GTRecord> iterator() {
+ count = 0;
+ return new Iterator<GTRecord>() {
+ GTRecord record = new GTRecord(info);
+ GTRecord next;
+ ByteBuffer buf = ByteBuffer.allocate(info.getMaxRecordLength());
+
+ @Override
+ public boolean hasNext() {
+ if (next != null)
+ return true;
+
+ try {
+ if (din.available() > 0) {
+ int len = din.readInt();
+ din.read(buf.array(), buf.arrayOffset(), len);
+ buf.clear();
+ buf.limit(len);
+ record.loadColumns(info.getAllColumns(), buf);
+ next = record;
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ return next != null;
+ }
+
+ @Override
+ public GTRecord next() {
+ if (next == null) {
+ hasNext();
+ if (next == null)
+ throw new NoSuchElementException();
+ }
+ GTRecord r = next;
+ next = null;
+ count++;
+ return r;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ };
}
@Override
- public void remove() {
- throw new UnsupportedOperationException();
+ public GTInfo getInfo() {
+ return info;
}
@Override
- public void close() throws IOException {
- synchronized (lock) {
- din.close();
- diskPart.closeRead();
- if (debug)
- logger.debug(MemDiskStore.this + " read end @ " + readOffset + ", " + (memRead) + " from mem, " + (diskRead) + " from disk, " + nReadCalls + " read() calls");
- }
+ public int getScannedRowCount() {
+ return count;
}
}
- private class Writer implements IGTStoreWriter {
+ private class Writer implements IGTWriter {
final DataOutputStream dout;
+ final ByteBuffer buf;
long writeOffset;
long memWrite = 0;
long diskWrite = 0;
@@ -273,6 +295,7 @@ public class MemDiskStore implements IGTStore, Closeable {
boolean closed = false;
Writer(long startOffset) throws IOException {
+ buf = ByteBuffer.allocate(info.getMaxRecordLength());
writeOffset = 0; // TODO does not support append yet
memPart.clear();
diskPart.clear();
@@ -294,7 +317,7 @@ public class MemDiskStore implements IGTStore, Closeable {
@Override
public void write(byte[] bytes, int offset, int length) throws IOException {
- // lock inside memPart.write() and diskPartm.write()
+ // lock inside memPart.write() and diskPart.write()
nWriteCalls++;
while (length > 0) {
int n;
@@ -318,8 +341,13 @@ public class MemDiskStore implements IGTStore, Closeable {
}
@Override
- public void write(GTRowBlock block) throws IOException {
- block.export(dout);
+ public void write(GTRecord rec) throws IOException {
+ buf.clear();
+ rec.exportColumns(info.getAllColumns(), buf);
+
+ int len = buf.position();
+ dout.writeInt(len);
+ dout.write(buf.array(), buf.arrayOffset(), len);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/cube/util/BitSetSerializer.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/util/BitSetSerializer.java b/core-cube/src/main/java/org/apache/kylin/cube/util/BitSetSerializer.java
new file mode 100644
index 0000000..9417fe4
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/cube/util/BitSetSerializer.java
@@ -0,0 +1,62 @@
+/*
+ * 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.util;
+
+import java.util.BitSet;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+public class BitSetSerializer extends Serializer<BitSet> {
+
+ @Override
+ public BitSet copy(final Kryo kryo, final BitSet original) {
+ final BitSet result = new BitSet();
+ final int length = original.length();
+ for (int i = 0; i < length; i++) {
+ result.set(i, original.get(i));
+ }
+ return result;
+ }
+
+ @Override
+ public void write(final Kryo kryo, final Output output, final BitSet bitSet) {
+ final int len = bitSet.length();
+
+ output.writeInt(len, true);
+
+ for (int i = 0; i < len; i++) {
+ output.writeBoolean(bitSet.get(i));
+ }
+ }
+
+ @Override
+ public BitSet read(final Kryo kryo, final Input input, final Class<BitSet> bitSetClass) {
+ final int len = input.readInt(true);
+ final BitSet ret = new BitSet(len);
+
+ for (int i = 0; i < len; i++) {
+ ret.set(i, input.readBoolean());
+ }
+
+ return ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/cube/util/KryoUtils.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/util/KryoUtils.java b/core-cube/src/main/java/org/apache/kylin/cube/util/KryoUtils.java
new file mode 100644
index 0000000..9dbe0d2
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/cube/util/KryoUtils.java
@@ -0,0 +1,61 @@
+/*
+ * 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.util;
+
+import java.util.BitSet;
+
+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.register(BitSet.class, new BitSetSerializer());
+ _Kryo.set(kryo);
+ }
+
+ return _Kryo.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java
index 4af4545..6950e98 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTAggregateScanner.java
@@ -56,11 +56,6 @@ public class GTAggregateScanner implements IGTScanner {
}
@Override
- public int getScannedRowBlockCount() {
- return inputScanner.getScannedRowBlockCount();
- }
-
- @Override
public void close() throws IOException {
inputScanner.close();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/GTBuilder.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTBuilder.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTBuilder.java
index 7308b5c..31ea9e2 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTBuilder.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTBuilder.java
@@ -1,22 +1,15 @@
package org.apache.kylin.gridtable;
import java.io.Closeable;
-import java.io.Flushable;
import java.io.IOException;
-import org.apache.kylin.gridtable.IGTStore.IGTStoreWriter;
-
-public class GTBuilder implements Closeable, Flushable {
+public class GTBuilder implements Closeable {
@SuppressWarnings("unused")
final private GTInfo info;
- final private IGTStoreWriter storeWriter;
-
- final private GTRowBlock block;
- final private GTRowBlock.Writer blockWriter;
+ final private IGTWriter storeWriter;
private int writtenRowCount;
- private int writtenRowBlockCount;
GTBuilder(GTInfo info, int shard, IGTStore store) throws IOException {
this(info, shard, store, false);
@@ -25,42 +18,20 @@ public class GTBuilder implements Closeable, Flushable {
GTBuilder(GTInfo info, int shard, IGTStore store, boolean append) throws IOException {
this.info = info;
- block = GTRowBlock.allocate(info);
- blockWriter = block.getWriter();
if (append) {
- storeWriter = store.append(shard, blockWriter);
- if (block.isFull()) {
- blockWriter.clearForNext();
- }
+ storeWriter = store.append(shard);
} else {
storeWriter = store.rebuild(shard);
}
}
public void write(GTRecord r) throws IOException {
- blockWriter.append(r);
+ storeWriter.write(r);
writtenRowCount++;
-
- if (block.isFull()) {
- flush();
- }
- }
-
- @Override
- public void flush() throws IOException {
- blockWriter.readyForFlush();
- storeWriter.write(block);
- writtenRowBlockCount++;
- if (block.isFull()) {
- blockWriter.clearForNext();
- }
}
@Override
public void close() throws IOException {
- if (block.isEmpty() == false) {
- flush();
- }
storeWriter.close();
}
@@ -68,7 +39,4 @@ public class GTBuilder implements Closeable, Flushable {
return writtenRowCount;
}
- public int getWrittenRowBlockCount() {
- return writtenRowBlockCount;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/GTFilterScanner.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTFilterScanner.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTFilterScanner.java
index 72dddce..bf4ce8c 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTFilterScanner.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/GTFilterScanner.java
@@ -43,11 +43,6 @@ public class GTFilterScanner implements IGTScanner {
}
@Override
- public int getScannedRowBlockCount() {
- return inputScanner.getScannedRowBlockCount();
- }
-
- @Override
public void close() throws IOException {
inputScanner.close();
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 b45a015..d4fe3fb 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
@@ -6,6 +6,8 @@ import java.util.Iterator;
import java.util.LinkedList;
import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.cube.gridtable.CubeCodeSystem;
+import org.apache.kylin.cube.util.KryoUtils;
import org.apache.kylin.metadata.model.DataType;
import org.apache.kylin.metadata.model.TblColRef;
@@ -39,16 +41,6 @@ public class GTInfo {
private GTInfo() {
}
- public IGTCodeSystem trimCodeSystemForCoprocessor() {
- IGTCodeSystem originalCodeSystem = this.codeSystem;
- codeSystem = this.codeSystem.trim();
- return originalCodeSystem;
- }
-
- public void setIGTCodeSystem(IGTCodeSystem codeSystem) {
- this.codeSystem = codeSystem;
- }
-
public String getTableName() {
return tableName;
}
@@ -64,6 +56,10 @@ public class GTInfo {
public ImmutableBitSet getPrimaryKey() {
return primaryKey;
}
+
+ public ImmutableBitSet getAllColumns() {
+ return colAll;
+ }
public boolean isShardingEnabled() {
return nShards > 0;
@@ -128,7 +124,6 @@ public class GTInfo {
}
void validate() {
-
if (codeSystem == null)
throw new IllegalStateException();
@@ -250,4 +245,19 @@ public class GTInfo {
}
}
+ public static byte[] serialize(GTInfo info) {
+ if (info.codeSystem instanceof CubeCodeSystem) {
+ CubeCodeSystem origin = (CubeCodeSystem) info.codeSystem;
+ info.codeSystem = origin.trimForCoprocessor();
+ byte[] trimmedInfoBytes = KryoUtils.serialize(info);
+ info.codeSystem = origin;
+ return trimmedInfoBytes;
+ } else {
+ return KryoUtils.serialize(info);
+ }
+ }
+
+ public static GTInfo deserialize(byte[] bytes) {
+ return KryoUtils.deserialize(bytes, GTInfo.class);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 e8bf69d..dbfdf57 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
@@ -23,9 +23,8 @@ public class GTRecord implements Comparable<GTRecord> {
public GTRecord(GTInfo info, ImmutableBitSet maskForEqualHashComp) {
this.cols = new ByteArray[info.getColumnCount()];
for (int i = 0; i < this.cols.length; i++) {
- if (maskForEqualHashComp.get(i)) {
- this.cols[i] = new ByteArray();
- }
+ // consider column projection by pass in another bit set
+ this.cols[i] = new ByteArray();
}
this.info = info;
this.maskForEqualHashComp = maskForEqualHashComp;
@@ -214,7 +213,7 @@ public class GTRecord implements Comparable<GTRecord> {
}
ByteArray buf = ByteArray.allocate(len);
- exportColumns(info.primaryKey, buf);
+ exportColumns(selectedCols, buf);
return buf;
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 99be3b7..18884f7 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,11 +37,6 @@ 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/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/GTStoreBridgeScanner.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/GTStoreBridgeScanner.java b/core-cube/src/main/java/org/apache/kylin/gridtable/GTStoreBridgeScanner.java
deleted file mode 100644
index d7d7f9d..0000000
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/GTStoreBridgeScanner.java
+++ /dev/null
@@ -1,111 +0,0 @@
-package org.apache.kylin.gridtable;
-
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-
-import org.apache.kylin.common.util.ImmutableBitSet;
-import org.apache.kylin.gridtable.IGTStore.IGTStoreScanner;
-
-public class GTStoreBridgeScanner implements IGTScanner {
-
- final GTInfo info;
- final IGTStoreScanner storeScanner;
- final ImmutableBitSet selectedColBlocks;
-
- private GTRowBlock.Reader curBlockReader;
- private GTRecord next;
- final private GTRecord oneRecord; // avoid instance creation
-
- private int scannedRowCount = 0;
- private int scannedRowBlockCount = 0;
-
- public GTStoreBridgeScanner(GTInfo info, GTScanRequest req, IGTStoreScanner storeScanner) throws IOException {
- this.info = info;
- this.selectedColBlocks = req.getSelectedColBlocks();
- this.storeScanner = storeScanner;
- this.oneRecord = new GTRecord(info);
- }
-
- @Override
- public GTInfo getInfo() {
- return info;
- }
-
- @Override
- public int getScannedRowCount() {
- return scannedRowCount;
- }
-
- @Override
- public int getScannedRowBlockCount() {
- return scannedRowBlockCount;
- }
-
- @Override
- public void close() throws IOException {
- storeScanner.close();
- }
-
- @Override
- public Iterator<GTRecord> iterator() {
- return new Iterator<GTRecord>() {
-
- @Override
- public boolean hasNext() {
- if (next != null)
- return true;
-
- if (fetchOneRecord()) {
- next = oneRecord;
- return true;
- } else {
- return false;
- }
- }
-
- private boolean fetchOneRecord() {
- while (true) {
- // get a block
- if (curBlockReader == null) {
- if (storeScanner.hasNext()) {
- curBlockReader = storeScanner.next().getReader(selectedColBlocks);
- scannedRowBlockCount++;
- } else {
- return false;
- }
- }
- // if block exhausted, try next block
- if (curBlockReader.hasNext() == false) {
- curBlockReader = null;
- continue;
- }
- // fetch a row
- curBlockReader.fetchNext(oneRecord);
- scannedRowCount++;
- return true;
- }
- }
-
- @Override
- public GTRecord next() {
- // fetch next record
- if (next == null) {
- hasNext();
- if (next == null)
- throw new NoSuchElementException();
- }
-
- GTRecord result = next;
- next = null;
- return result;
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
- };
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 be79681..48bda9f 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,8 +9,6 @@ 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/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/IGTScanner.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTScanner.java b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTScanner.java
index 7b95b91..a82dffa 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTScanner.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTScanner.java
@@ -8,6 +8,4 @@ public interface IGTScanner extends Iterable<GTRecord>, Closeable {
int getScannedRowCount();
- int getScannedRowBlockCount();
-
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/IGTStore.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTStore.java b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTStore.java
index dcc6098..5282544 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTStore.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTStore.java
@@ -1,24 +1,15 @@
package org.apache.kylin.gridtable;
-import java.io.Closeable;
import java.io.IOException;
-import java.util.Iterator;
public interface IGTStore {
GTInfo getInfo();
- IGTStoreWriter rebuild(int shard) throws IOException;
+ IGTWriter rebuild(int shard) throws IOException;
- IGTStoreWriter append(int shard, GTRowBlock.Writer fillLast) throws IOException;
+ IGTWriter append(int shard) throws IOException;
IGTScanner scan(GTScanRequest scanRequest) throws IOException;
- interface IGTStoreWriter extends Closeable {
- void write(GTRowBlock block) throws IOException;
- }
-
- interface IGTStoreScanner extends Iterator<GTRowBlock>, Closeable {
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/IGTWriter.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/IGTWriter.java b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTWriter.java
new file mode 100644
index 0000000..6733c68
--- /dev/null
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/IGTWriter.java
@@ -0,0 +1,9 @@
+package org.apache.kylin.gridtable;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+public interface IGTWriter extends Closeable {
+
+ void write(GTRecord rec) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/main/java/org/apache/kylin/gridtable/memstore/GTSimpleMemStore.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/gridtable/memstore/GTSimpleMemStore.java b/core-cube/src/main/java/org/apache/kylin/gridtable/memstore/GTSimpleMemStore.java
index 9eba771..d7074e4 100644
--- a/core-cube/src/main/java/org/apache/kylin/gridtable/memstore/GTSimpleMemStore.java
+++ b/core-cube/src/main/java/org/apache/kylin/gridtable/memstore/GTSimpleMemStore.java
@@ -1,25 +1,27 @@
package org.apache.kylin.gridtable.memstore;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.gridtable.GTInfo;
-import org.apache.kylin.gridtable.GTRowBlock;
+import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRequest;
-import org.apache.kylin.gridtable.GTStoreBridgeScanner;
import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.gridtable.IGTStore;
+import org.apache.kylin.gridtable.IGTWriter;
public class GTSimpleMemStore implements IGTStore {
final GTInfo info;
- final List<GTRowBlock> rowBlockList;
+ final List<byte[]> rowList;
public GTSimpleMemStore(GTInfo info) {
this.info = info;
- this.rowBlockList = new ArrayList<GTRowBlock>();
+ this.rowList = new ArrayList<byte[]>();
if (info.isShardingEnabled())
throw new UnsupportedOperationException();
@@ -31,87 +33,90 @@ public class GTSimpleMemStore implements IGTStore {
}
public long memoryUsage() {
- if (rowBlockList.size() == 0) {
- return 0;
- } else {
- return rowBlockList.get(0).exportLength() * Long.valueOf(rowBlockList.size());
+ long sum = 0;
+ for (byte[] bytes : rowList) {
+ sum += bytes.length;
}
+ return sum;
}
@Override
- public IGTStoreWriter rebuild(int shard) {
- rowBlockList.clear();
- return new Writer(rowBlockList);
+ public IGTWriter rebuild(int shard) {
+ rowList.clear();
+ return new Writer();
}
@Override
- public IGTStoreWriter append(int shard, GTRowBlock.Writer fillLast) {
- if (rowBlockList.size() > 0) {
- GTRowBlock last = rowBlockList.get(rowBlockList.size() - 1);
- fillLast.copyFrom(last);
- }
- return new Writer(rowBlockList);
+ public IGTWriter append(int shard) {
+ return new Writer();
}
- private static class Writer implements IGTStoreWriter {
-
- private final List<GTRowBlock> rowBlockList;
-
- Writer(List<GTRowBlock> rowBlockList) {
- this.rowBlockList = rowBlockList;
- }
-
+ private class Writer implements IGTWriter {
@Override
- public void close() throws IOException {
+ public void write(GTRecord r) throws IOException {
+ ByteArray byteArray = r.exportColumns(info.getAllColumns()).copy();
+ assert byteArray.offset() == 0;
+ assert byteArray.array().length == byteArray.length();
+ rowList.add(byteArray.array());
}
@Override
- public void write(GTRowBlock block) throws IOException {
- GTRowBlock copy = block.copy();
- int id = block.getSequenceId();
- if (id < rowBlockList.size()) {
- rowBlockList.set(id, copy);
- } else {
- assert id == rowBlockList.size();
- rowBlockList.add(copy);
- }
+ public void close() throws IOException {
}
}
@Override
- public IGTScanner scan(GTScanRequest scanRequest) throws IOException {
- return new GTStoreBridgeScanner(info, scanRequest, oldScan(scanRequest));
- }
+ public IGTScanner scan(GTScanRequest scanRequest) {
- private IGTStoreScanner oldScan(GTScanRequest scanRequest) {
-
- return new IGTStoreScanner() {
- Iterator<GTRowBlock> it = rowBlockList.iterator();
+ return new IGTScanner() {
+ int count;
@Override
- public boolean hasNext() {
- return it.hasNext();
+ public GTInfo getInfo() {
+ return info;
}
@Override
- public GTRowBlock next() {
- return it.next();
+ public int getScannedRowCount() {
+ return count;
}
@Override
- public void remove() {
- throw new UnsupportedOperationException();
+ public void close() throws IOException {
}
@Override
- public void close() throws IOException {
+ public Iterator<GTRecord> iterator() {
+ count = 0;
+ return new Iterator<GTRecord>() {
+ Iterator<byte[]> it = rowList.iterator();
+ GTRecord oneRecord = new GTRecord(info);
+
+ @Override
+ public boolean hasNext() {
+ return it.hasNext();
+ }
+
+ @Override
+ public GTRecord next() {
+ byte[] bytes = it.next();
+ oneRecord.loadColumns(info.getAllColumns(), ByteBuffer.wrap(bytes));
+ count++;
+ return oneRecord;
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
}
};
}
public void drop() throws IOException {
//will there be any concurrent issue? If yes, ArrayList should be replaced
- rowBlockList.clear();
+ rowList.clear();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/core-cube/src/test/java/org/apache/kylin/gridtable/SimpleGridTableTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/gridtable/SimpleGridTableTest.java b/core-cube/src/test/java/org/apache/kylin/gridtable/SimpleGridTableTest.java
index 01a1e55..6395a3b 100644
--- a/core-cube/src/test/java/org/apache/kylin/gridtable/SimpleGridTableTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/gridtable/SimpleGridTableTest.java
@@ -41,7 +41,6 @@ public class SimpleGridTableTest {
GTBuilder builder = rebuild(table);
IGTScanner scanner = scan(table);
- assertEquals(builder.getWrittenRowBlockCount(), scanner.getScannedRowBlockCount());
assertEquals(builder.getWrittenRowCount(), scanner.getScannedRowCount());
}
@@ -53,7 +52,6 @@ public class SimpleGridTableTest {
GTBuilder builder = rebuild(table);
IGTScanner scanner = scan(table);
- assertEquals(builder.getWrittenRowBlockCount(), scanner.getScannedRowBlockCount());
assertEquals(builder.getWrittenRowCount(), scanner.getScannedRowCount());
}
@@ -65,7 +63,6 @@ public class SimpleGridTableTest {
GTBuilder builder = rebuild(table);
IGTScanner scanner = scanAndAggregate(table);
- assertEquals(builder.getWrittenRowBlockCount(), scanner.getScannedRowBlockCount());
assertEquals(builder.getWrittenRowCount(), scanner.getScannedRowCount());
}
@@ -77,7 +74,6 @@ public class SimpleGridTableTest {
rebuildViaAppend(table);
IGTScanner scanner = scan(table);
- assertEquals(3, scanner.getScannedRowBlockCount());
assertEquals(10, scanner.getScannedRowCount());
}
@@ -93,7 +89,6 @@ public class SimpleGridTableTest {
System.out.println(r);
}
scanner.close();
- System.out.println("Scanned Row Block Count: " + scanner.getScannedRowBlockCount());
System.out.println("Scanned Row Count: " + scanner.getScannedRowCount());
return scanner;
}
@@ -128,7 +123,6 @@ public class SimpleGridTableTest {
System.out.println(r);
}
scanner.close();
- System.out.println("Scanned Row Block Count: " + scanner.getScannedRowBlockCount());
System.out.println("Scanned Row Count: " + scanner.getScannedRowCount());
return scanner;
}
@@ -140,7 +134,6 @@ public class SimpleGridTableTest {
}
builder.close();
- System.out.println("Written Row Block Count: " + builder.getWrittenRowBlockCount());
System.out.println("Written Row Count: " + builder.getWrittenRowCount());
return builder;
}
@@ -156,7 +149,6 @@ public class SimpleGridTableTest {
builder.write(data.get(i++));
builder.write(data.get(i++));
builder.close();
- System.out.println("Written Row Block Count: " + builder.getWrittenRowBlockCount());
System.out.println("Written Row Count: " + builder.getWrittenRowCount());
builder = table.append();
@@ -164,20 +156,17 @@ public class SimpleGridTableTest {
builder.write(data.get(i++));
builder.write(data.get(i++));
builder.close();
- System.out.println("Written Row Block Count: " + builder.getWrittenRowBlockCount());
System.out.println("Written Row Count: " + builder.getWrittenRowCount());
builder = table.append();
builder.write(data.get(i++));
builder.write(data.get(i++));
builder.close();
- System.out.println("Written Row Block Count: " + builder.getWrittenRowBlockCount());
System.out.println("Written Row Count: " + builder.getWrittenRowCount());
builder = table.append();
builder.write(data.get(i++));
builder.close();
- System.out.println("Written Row Block Count: " + builder.getWrittenRowBlockCount());
System.out.println("Written Row Count: " + builder.getWrittenRowCount());
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/storage/pom.xml
----------------------------------------------------------------------
diff --git a/storage/pom.xml b/storage/pom.xml
index f9f7d78..9b11484 100644
--- a/storage/pom.xml
+++ b/storage/pom.xml
@@ -40,10 +40,6 @@
<artifactId>kylin-storage-hbase</artifactId>
<version>${project.parent.version}</version>
</dependency>
- <dependency>
- <groupId>com.esotericsoftware</groupId>
- <artifactId>kryo-shaded</artifactId>
- </dependency>
<!-- Env & Test -->
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 2057e0b..ac80b61 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
@@ -38,6 +38,7 @@ import org.apache.kylin.common.util.ImmutableBitSet;
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.util.KryoUtils;
import org.apache.kylin.gridtable.GTInfo;
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRequest;
@@ -47,7 +48,6 @@ import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitPr
import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitRequest.Builder;
import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitResponse;
import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitService;
-import org.apache.kylin.storage.util.KryoUtils;
import com.google.common.base.Function;
import com.google.common.collect.Collections2;
@@ -60,9 +60,6 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
private GTInfo info;
private Iterator<byte[]> blocks;
- private int scannedRowCount = 0;
- private int scannedRowBlockCount = 0;
-
public EndpintResultsAsGTScanner(GTInfo info, Iterator<byte[]> blocks) {
this.info = info;
this.blocks = blocks;
@@ -79,11 +76,6 @@ public class CubeHBaseEndpointRPC extends CubeHBaseRPC {
}
@Override
- public int getScannedRowBlockCount() {
- return 0;
- }
-
- @Override
public void close() throws IOException {
//do nothing
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 53438e4..ce0ef55 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
@@ -70,7 +70,7 @@ public class CubeHBaseScanRPC extends CubeHBaseRPC {
}
};
- IGTStore store = new HBaseGTStore(cellListIterator, scanRequest, hbaseColumns);
+ IGTStore store = new HBaseReadonlyStore(cellListIterator, scanRequest, hbaseColumns);
IGTScanner rawScanner = store.scan(scanRequest);
return scanRequest.decorateScanner(rawScanner);
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/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 71dea62..8368902 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
@@ -22,12 +22,10 @@ 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.IGTCodeSystem;
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 org.apache.kylin.storage.util.KryoUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -65,10 +63,8 @@ public class CubeScanner implements IGTScanner {
scanRequests = Lists.newArrayListWithCapacity(scanRanges.size());
- IGTCodeSystem origin = info.trimCodeSystemForCoprocessor();
- trimmedInfoBytes = KryoUtils.serialize(info);
- info.setIGTCodeSystem(origin);
- GTInfo trimmedInfo = KryoUtils.deserialize(this.trimmedInfoBytes, GTInfo.class);
+ trimmedInfoBytes = GTInfo.serialize(info);
+ GTInfo trimmedInfo = GTInfo.deserialize(trimmedInfoBytes);
for (GTScanRange range : scanRanges) {
scanRequests.add(new GTScanRequest(trimmedInfo, range, gtDimensions, gtAggrGroups, gtAggrMetrics, gtAggrFuncs, gtFilter, allowPreAggregate));
@@ -154,11 +150,6 @@ public class CubeScanner implements IGTScanner {
return scanner.getScannedRowCount();
}
- @Override
- public int getScannedRowBlockCount() {
- return scanner.getScannedRowBlockCount();
- }
-
static class RemoteGTRecordAdapter implements Iterable<GTRecord> {
private final GTInfo info;
@@ -269,17 +260,6 @@ public class CubeScanner implements IGTScanner {
return result;
}
- public int getScannedRowBlockCount() {
- int result = 0;
- for (int i = 0; i < inputScanners.length; i++) {
- if (inputScanners[i] == null)
- break;
-
- result += inputScanners[i].getScannedRowBlockCount();
- }
- return result;
- }
-
}
}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java b/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java
deleted file mode 100644
index 1047569..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/cube/HBaseGTStore.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kylin.storage.cube;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.kylin.common.util.ImmutableBitSet;
-import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.gridtable.GTInfo;
-import org.apache.kylin.gridtable.GTRowBlock;
-import org.apache.kylin.gridtable.GTScanRequest;
-import org.apache.kylin.gridtable.GTStoreBridgeScanner;
-import org.apache.kylin.gridtable.IGTScanner;
-import org.apache.kylin.gridtable.IGTStore;
-
-public class HBaseGTStore implements IGTStore {
-
- private CellListIterator cellListIterator;
-
- private GTInfo info;
- private List<Pair<byte[], byte[]>> hbaseColumns;
- private ImmutableBitSet selectedColBlocks;
-
- private GTRowBlock oneBlock;
-
- public HBaseGTStore(CellListIterator cellListIterator, GTScanRequest gtScanRequest, List<Pair<byte[], byte[]>> hbaseColumns) {
- this.cellListIterator = cellListIterator;
-
- this.info = gtScanRequest.getInfo();
- this.hbaseColumns = hbaseColumns;
- this.selectedColBlocks = gtScanRequest.getSelectedColBlocks().set(0);
-
- oneBlock = new GTRowBlock(info); // avoid object creation
- oneBlock.setNumberOfRows(1); // row block is always disabled for cubes, row block contains only one record
- }
-
- @Override
- public GTInfo getInfo() {
- return info;
- }
-
- @Override
- public IGTStoreWriter rebuild(int shard) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public IGTStoreWriter append(int shard, GTRowBlock.Writer fillLast) throws IOException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public IGTScanner scan(GTScanRequest scanRequest) throws IOException {
- return new GTStoreBridgeScanner(info, scanRequest, oldScan(scanRequest));
- }
-
- private IGTStoreScanner oldScan(GTScanRequest scanRequest) throws IOException {
- return new IGTStoreScanner() {
- @Override
- public boolean hasNext() {
- return cellListIterator.hasNext();
- }
-
- @Override
- public GTRowBlock next() {
- List<Cell> oneRow = cellListIterator.next();
- if (oneRow.size() < 1) {
- throw new IllegalStateException("cell list's size less than 1");
- }
-
- // dimensions, set to primary key, also the 0th column block
- Cell firstCell = oneRow.get(0);
- byte[] rowkey = firstCell.getRowArray();
- oneBlock.getPrimaryKey().set(rowkey, RowConstants.ROWKEY_CUBOIDID_LEN + firstCell.getRowOffset(), firstCell.getRowLength() - RowConstants.ROWKEY_CUBOIDID_LEN);
- oneBlock.getCellBlock(0).set(rowkey, RowConstants.ROWKEY_CUBOIDID_LEN + firstCell.getRowOffset(), firstCell.getRowLength() - RowConstants.ROWKEY_CUBOIDID_LEN);
-
- // metrics
- int hbaseColIdx = 0;
- for (int i = 1; i < selectedColBlocks.trueBitCount(); i++) {
- int colBlockIdx = selectedColBlocks.trueBitAt(i);
- Pair<byte[], byte[]> hbaseColumn = hbaseColumns.get(hbaseColIdx++);
- Cell cell = CubeHBaseRPC.findCell(oneRow, hbaseColumn.getFirst(), hbaseColumn.getSecond());
- oneBlock.getCellBlock(colBlockIdx).set(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
- }
- return oneBlock;
-
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void close() throws IOException {
- cellListIterator.close();
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/storage/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java b/storage/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java
new file mode 100644
index 0000000..096dba7
--- /dev/null
+++ b/storage/src/main/java/org/apache/kylin/storage/cube/HBaseReadonlyStore.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kylin.storage.cube;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.kylin.common.util.ImmutableBitSet;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.gridtable.GTInfo;
+import org.apache.kylin.gridtable.GTRecord;
+import org.apache.kylin.gridtable.GTScanRequest;
+import org.apache.kylin.gridtable.IGTScanner;
+import org.apache.kylin.gridtable.IGTStore;
+import org.apache.kylin.gridtable.IGTWriter;
+
+public class HBaseReadonlyStore implements IGTStore {
+
+ private CellListIterator cellListIterator;
+
+ private GTInfo info;
+ private List<Pair<byte[], byte[]>> hbaseColumns;
+ private ImmutableBitSet selectedColBlocks;
+
+ public HBaseReadonlyStore(CellListIterator cellListIterator, GTScanRequest gtScanRequest, List<Pair<byte[], byte[]>> hbaseColumns) {
+ this.cellListIterator = cellListIterator;
+
+ this.info = gtScanRequest.getInfo();
+ this.hbaseColumns = hbaseColumns;
+ this.selectedColBlocks = gtScanRequest.getSelectedColBlocks().set(0);
+ }
+
+ @Override
+ public GTInfo getInfo() {
+ return info;
+ }
+
+ @Override
+ public IGTWriter rebuild(int shard) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public IGTWriter append(int shard) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public IGTScanner scan(GTScanRequest scanRequest) throws IOException {
+ return new IGTScanner() {
+ int count;
+
+ @Override
+ public void close() throws IOException {
+ cellListIterator.close();
+ }
+
+ @Override
+ public Iterator<GTRecord> iterator() {
+ return new Iterator<GTRecord>() {
+ GTRecord oneRecord = new GTRecord(info); // avoid object creation
+
+ @Override
+ public boolean hasNext() {
+ return cellListIterator.hasNext();
+ }
+
+ @Override
+ public GTRecord next() {
+ List<Cell> oneRow = cellListIterator.next();
+ if (oneRow.size() < 1) {
+ throw new IllegalStateException("cell list's size less than 1");
+ }
+
+ ByteBuffer buf;
+
+ // dimensions, set to primary key, also the 0th column block
+ Cell firstCell = oneRow.get(0);
+ buf = byteBuffer(firstCell.getRowArray(), RowConstants.ROWKEY_CUBOIDID_LEN + firstCell.getRowOffset(), firstCell.getRowLength() - RowConstants.ROWKEY_CUBOIDID_LEN);
+ oneRecord.loadCellBlock(0, buf);
+
+ // metrics
+ int hbaseColIdx = 0;
+ for (int i = 1; i < selectedColBlocks.trueBitCount(); i++) {
+ int colBlockIdx = selectedColBlocks.trueBitAt(i);
+ Pair<byte[], byte[]> hbaseColumn = hbaseColumns.get(hbaseColIdx++);
+ Cell cell = CubeHBaseRPC.findCell(oneRow, hbaseColumn.getFirst(), hbaseColumn.getSecond());
+ buf = byteBuffer(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ oneRecord.loadCellBlock(colBlockIdx, buf);
+ }
+ return oneRecord;
+
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ private ByteBuffer byteBuffer(byte[] array, int offset, int length) {
+ return ByteBuffer.wrap(array, offset, length);
+ }
+
+ };
+ }
+
+ @Override
+ public GTInfo getInfo() {
+ return info;
+ }
+
+ @Override
+ public int getScannedRowCount() {
+ return count;
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java
index 300469a..c44ac52 100644
--- a/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java
+++ b/storage/src/main/java/org/apache/kylin/storage/hbase/coprocessor/endpoint/CubeVisitService.java
@@ -36,18 +36,18 @@ import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.kylin.common.util.CompressionUtils;
import org.apache.kylin.cube.kv.RowConstants;
+import org.apache.kylin.cube.util.KryoUtils;
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRequest;
import org.apache.kylin.gridtable.IGTScanner;
import org.apache.kylin.gridtable.IGTStore;
import org.apache.kylin.storage.cube.CellListIterator;
import org.apache.kylin.storage.cube.CubeHBaseRPC;
-import org.apache.kylin.storage.cube.HBaseGTStore;
+import org.apache.kylin.storage.cube.HBaseReadonlyStore;
import org.apache.kylin.storage.cube.RawScan;
import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos;
import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitResponse.Builder;
import org.apache.kylin.storage.hbase.coprocessor.endpoint.generated.CubeVisitProtos.CubeVisitResponse.Stats;
-import org.apache.kylin.storage.util.KryoUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -138,7 +138,7 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
innerScanner = region.getScanner(scan);
InnerScannerAsIterator cellListIterator = new InnerScannerAsIterator(innerScanner);
- IGTStore store = new HBaseGTStore(cellListIterator, scanReq, hbaseRawScan.hbaseColumns);
+ IGTStore store = new HBaseReadonlyStore(cellListIterator, scanReq, hbaseRawScan.hbaseColumns);
IGTScanner rawScanner = store.scan(scanReq);
IGTScanner finalScanner = scanReq.decorateScanner(rawScanner);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/storage/src/main/java/org/apache/kylin/storage/util/BitSetSerializer.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/util/BitSetSerializer.java b/storage/src/main/java/org/apache/kylin/storage/util/BitSetSerializer.java
deleted file mode 100644
index 77e3e26..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/util/BitSetSerializer.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package org.apache.kylin.storage.util;
-
-import java.util.BitSet;
-
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.Serializer;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
-
-public class BitSetSerializer extends Serializer<BitSet> {
-
- @Override
- public BitSet copy(final Kryo kryo, final BitSet original) {
- final BitSet result = new BitSet();
- final int length = original.length();
- for (int i = 0; i < length; i++) {
- result.set(i, original.get(i));
- }
- return result;
- }
-
- @Override
- public void write(final Kryo kryo, final Output output, final BitSet bitSet) {
- final int len = bitSet.length();
-
- output.writeInt(len, true);
-
- for (int i = 0; i < len; i++) {
- output.writeBoolean(bitSet.get(i));
- }
- }
-
- @Override
- public BitSet read(final Kryo kryo, final Input input, final Class<BitSet> bitSetClass) {
- final int len = input.readInt(true);
- final BitSet ret = new BitSet(len);
-
- for (int i = 0; i < len; i++) {
- ret.set(i, input.readBoolean());
- }
-
- return ret;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/4741004c/storage/src/main/java/org/apache/kylin/storage/util/KryoUtils.java
----------------------------------------------------------------------
diff --git a/storage/src/main/java/org/apache/kylin/storage/util/KryoUtils.java b/storage/src/main/java/org/apache/kylin/storage/util/KryoUtils.java
deleted file mode 100644
index 3f67fa0..0000000
--- a/storage/src/main/java/org/apache/kylin/storage/util/KryoUtils.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package org.apache.kylin.storage.util;
-
-import java.util.BitSet;
-
-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.register(BitSet.class, new BitSetSerializer());
- _Kryo.set(kryo);
- }
-
- return _Kryo.get();
- }
-}