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 2016/05/22 13:24:03 UTC
kylin git commit: KYLIN-1718 Grow ByteBuffer Dynamically in Cube
Building and Query
Repository: kylin
Updated Branches:
refs/heads/master f2f3d6866 -> d56abdea8
KYLIN-1718 Grow ByteBuffer Dynamically in Cube Building and Query
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/d56abdea
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/d56abdea
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/d56abdea
Branch: refs/heads/master
Commit: d56abdea858930be1df9e1772e323fd9cda31289
Parents: f2f3d68
Author: Yang Li <li...@apache.org>
Authored: Sun May 22 21:23:47 2016 +0800
Committer: Yang Li <li...@apache.org>
Committed: Sun May 22 21:23:47 2016 +0800
----------------------------------------------------------------------
.../org/apache/kylin/cube/kv/RowConstants.java | 1 -
.../kylin/gridtable/GTAggregateScanner.java | 32 +++---
.../org/apache/kylin/gridtable/GTRecord.java | 9 ++
.../metadata/measure/MeasureCodecTest.java | 11 +-
.../kylin/measure/BufferedMeasureEncoder.java | 107 +++++++++++++++++++
.../org/apache/kylin/measure/MeasureCodec.java | 92 ----------------
.../apache/kylin/measure/MeasureDecoder.java | 86 +++++++++++++++
.../org/apache/kylin/measure/MeasureType.java | 1 +
.../kylin/engine/mr/common/CuboidStatsUtil.java | 3 +-
.../engine/mr/steps/BaseCuboidMapperBase.java | 15 ++-
.../kylin/engine/mr/steps/CuboidReducer.java | 11 +-
.../mr/steps/FactDistinctHiveColumnsMapper.java | 3 +-
.../engine/mr/steps/InMemCuboidReducer.java | 18 +---
.../kylin/engine/mr/steps/KVGTRecordWriter.java | 26 +++--
.../engine/mr/steps/MergeCuboidMapper.java | 11 +-
.../kylin/engine/mr/steps/CubeReducerTest.java | 18 ++--
.../apache/kylin/engine/spark/SparkCubing.java | 13 +--
.../spark/cube/DefaultTupleConverter.java | 28 ++---
.../cardinality/ColumnCardinalityMapper.java | 3 +-
.../cardinality/ColumnCardinalityReducer.java | 3 +-
.../ColumnCardinalityReducerTest.java | 3 +-
.../observer/ObserverAggregators.java | 13 +--
.../coprocessor/endpoint/CubeVisitService.java | 16 ++-
.../storage/hbase/steps/CubeHFileMapper.java | 6 +-
.../storage/hbase/steps/KeyValueCreator.java | 11 +-
.../storage/hbase/steps/RowValueDecoder.java | 6 +-
.../observer/AggregateRegionObserverTest.java | 8 +-
.../hbase/steps/CubeHFileMapper2Test.java | 7 +-
.../hbase/steps/RowValueDecoderTest.java | 15 ++-
29 files changed, 332 insertions(+), 244 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
index 987fb55..809e0a3 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/kv/RowConstants.java
@@ -43,7 +43,6 @@ public class RowConstants {
public static final byte[] ROWVALUE_DELIMITER_BYTES = { 7 };
public static final int ROWKEY_BUFFER_SIZE = 65 * 256;// a little more than 64 dimensions * 256 bytes each
- public static final int ROWVALUE_BUFFER_SIZE = 1024 * 1024; // 1 MB
// marker class
public static final byte[][] BYTE_ARR_MARKER = new byte[0][];
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/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 7356e77..a2ca40c 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
@@ -39,9 +39,9 @@ import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.common.util.MemoryBudgetController;
import org.apache.kylin.common.util.MemoryBudgetController.MemoryWaterLevel;
import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureAggregator;
import org.apache.kylin.measure.MeasureAggregators;
-import org.apache.kylin.measure.MeasureCodec;
import org.apache.kylin.metadata.datatype.DataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -153,7 +153,7 @@ public class GTAggregateScanner implements IGTScanner {
final List<Dump> dumps;
final int keyLength;
final boolean[] compareMask;
- final MeasureCodec measureCodec;
+ final BufferedMeasureEncoder measureCodec;
final Comparator<byte[]> bytesComparator = new Comparator<byte[]>() {
@Override
@@ -187,12 +187,15 @@ public class GTAggregateScanner implements IGTScanner {
measureCodec = createMeasureCodec();
}
- private MeasureCodec createMeasureCodec() {
+ private BufferedMeasureEncoder createMeasureCodec() {
DataType[] types = new DataType[metrics.trueBitCount()];
for (int i = 0; i < types.length; i++) {
types[i] = info.getColumnType(metrics.trueBitAt(i));
}
- return new MeasureCodec(types);
+
+ BufferedMeasureEncoder result = new BufferedMeasureEncoder(types);
+ result.setBufferSize(info.getMaxColumnLength(metrics));
+ return result;
}
private boolean[] createCompareMask() {
@@ -358,7 +361,7 @@ public class GTAggregateScanner implements IGTScanner {
class ReturningRecord {
final GTRecord record = new GTRecord(info);
- final ByteBuffer metricsBuf = ByteBuffer.allocate(info.getMaxColumnLength(metrics));
+ final Object[] tmpValues = new Object[metrics.trueBitCount()];
void load(byte[] key, MeasureAggregator[] value) {
int offset = 0;
@@ -368,12 +371,18 @@ public class GTAggregateScanner implements IGTScanner {
record.cols[c].set(key, offset, columnLength);
offset += columnLength;
}
- metricsBuf.clear();
+
+ for (int i = 0; i < value.length; i++) {
+ tmpValues[i] = value[i].getState();
+ }
+
+ byte[] bytes = measureCodec.encode(tmpValues).array();
+ int[] sizes = measureCodec.getMeasureSizes();
+ offset = 0;
for (int i = 0; i < value.length; i++) {
int col = metrics.trueBitAt(i);
- int pos = metricsBuf.position();
- info.codeSystem.encodeColumnValue(col, value[i].getState(), metricsBuf);
- record.cols[col].set(metricsBuf.array(), pos, metricsBuf.position() - pos);
+ record.cols[col].set(bytes, offset, sizes[i]);
+ offset += sizes[i];
}
}
}
@@ -430,7 +439,6 @@ public class GTAggregateScanner implements IGTScanner {
if (buffMap != null) {
ObjectOutputStream oos = null;
Object[] aggrResult = null;
- final ByteBuffer metricsBuf = ByteBuffer.allocate(info.getMaxColumnLength(metrics));
try {
dumpedFile = File.createTempFile("KYLIN_AGGR_", ".tmp");
@@ -438,12 +446,10 @@ public class GTAggregateScanner implements IGTScanner {
oos = new ObjectOutputStream(new FileOutputStream(dumpedFile));
oos.writeInt(buffMap.size());
for (Entry<byte[], MeasureAggregator[]> entry : buffMap.entrySet()) {
- metricsBuf.clear();
-
MeasureAggregators aggs = new MeasureAggregators(entry.getValue());
aggrResult = new Object[metrics.trueBitCount()];
aggs.collectStates(aggrResult);
- measureCodec.encode(aggrResult, metricsBuf);
+ ByteBuffer metricsBuf = measureCodec.encode(aggrResult);
oos.writeObject(entry.getKey());
oos.writeObject(metricsBuf.array());
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/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 bccd0c5..f3cfc6a 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
@@ -136,6 +136,15 @@ public class GTRecord implements Comparable<GTRecord> {
}
return result;
}
+
+ public int sizeOf(ImmutableBitSet selectedCols) {
+ int size = 0;
+ for (int i = 0; i < selectedCols.trueBitCount(); i++) {
+ int c = selectedCols.trueBitAt(i);
+ size += cols[c].length();
+ }
+ return size;
+ }
public GTRecord copy() {
return copy(info.colAll);
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/core-cube/src/test/java/org/apache/kylin/metadata/measure/MeasureCodecTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/metadata/measure/MeasureCodecTest.java b/core-cube/src/test/java/org/apache/kylin/metadata/measure/MeasureCodecTest.java
index 840e174..3f728b9 100644
--- a/core-cube/src/test/java/org/apache/kylin/metadata/measure/MeasureCodecTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/metadata/measure/MeasureCodecTest.java
@@ -23,10 +23,9 @@ import static org.junit.Assert.*;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
-import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
-import org.apache.kylin.cube.kv.RowConstants;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.bitmap.BitmapCounter;
+import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
import org.apache.kylin.metadata.datatype.DoubleMutable;
import org.apache.kylin.metadata.datatype.LongMutable;
import org.apache.kylin.metadata.model.FunctionDesc;
@@ -42,7 +41,7 @@ public class MeasureCodecTest {
public void basicTest() {
MeasureDesc descs[] = new MeasureDesc[] { measure("double"), measure("long"), measure
("decimal"), measure("HLLC16"), measure("bitmap") };
- MeasureCodec codec = new MeasureCodec(descs);
+ BufferedMeasureEncoder codec = new BufferedMeasureEncoder(descs);
DoubleMutable d = new DoubleMutable(1.0);
LongMutable l = new LongMutable(2);
@@ -56,9 +55,7 @@ public class MeasureCodecTest {
bitmap.add(Integer.MAX_VALUE-10);
Object values[] = new Object[] { d, l, b, hllc, bitmap };
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
-
- codec.encode(values, buf);
+ ByteBuffer buf = codec.encode(values);
buf.flip();
System.out.println("size: " + buf.limit());
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureEncoder.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureEncoder.java b/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureEncoder.java
new file mode 100644
index 0000000..88c7949
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureEncoder.java
@@ -0,0 +1,107 @@
+/*
+ * 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.measure;
+
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.model.MeasureDesc;
+
+/**
+ * This class embeds a reusable byte buffer for measure encoding, and is not thread-safe.
+ * The buffer will grow to accommodate BufferOverflowException until a limit.
+ * The problem here to solve is some measure type cannot provide accurate DataTypeSerializer.maxLength()
+ */
+@SuppressWarnings({ "unchecked" })
+public class BufferedMeasureEncoder {
+ public static final int DEFAULT_BUFFER_SIZE = 1024 * 1024; // 1 MB
+ public static final int MAX_BUFFER_SIZE = 64 * DEFAULT_BUFFER_SIZE; // 64 MB
+
+ final private MeasureDecoder codec;
+
+ private ByteBuffer buf;
+ final private int[] measureSizes;
+
+ public BufferedMeasureEncoder(Collection<MeasureDesc> measureDescs) {
+ this.codec = new MeasureDecoder(measureDescs);
+ this.measureSizes = new int[codec.nMeasures];
+ }
+
+ public BufferedMeasureEncoder(MeasureDesc... measureDescs) {
+ this.codec = new MeasureDecoder(measureDescs);
+ this.measureSizes = new int[codec.nMeasures];
+ }
+
+ public BufferedMeasureEncoder(DataType... dataTypes) {
+ this.codec = new MeasureDecoder(dataTypes);
+ this.measureSizes = new int[codec.nMeasures];
+ }
+
+ public BufferedMeasureEncoder(String... dataTypes) {
+ this.codec = new MeasureDecoder(dataTypes);
+ this.measureSizes = new int[codec.nMeasures];
+ }
+
+ /** return the buffer that contains result of last encoding */
+ public ByteBuffer getBuffer() {
+ return buf;
+ }
+
+ /** return the measure sizes of last encoding */
+ public int[] getMeasureSizes() {
+ return measureSizes;
+ }
+
+ public void setBufferSize(int size) {
+ buf = null; // release memory for GC
+ buf = ByteBuffer.allocate(size);
+ }
+
+ public void decode(ByteBuffer buf, Object[] result) {
+ codec.decode(buf, result);
+ }
+
+ public ByteBuffer encode(Object[] values) {
+ if (buf == null) {
+ setBufferSize(DEFAULT_BUFFER_SIZE);
+ }
+
+ assert values.length == codec.nMeasures;
+
+ while (true) {
+ try {
+ buf.clear();
+ for (int i = 0, pos = 0; i < codec.nMeasures; i++) {
+ codec.serializers[i].serialize(values[i], buf);
+ measureSizes[i] = buf.position() - pos;
+ pos = buf.position();
+ }
+ return buf;
+
+ } catch (BufferOverflowException boe) {
+ if (buf.capacity() >= MAX_BUFFER_SIZE)
+ throw boe;
+
+ setBufferSize(buf.capacity() * 2);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java
deleted file mode 100644
index 2794fa8..0000000
--- a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java
+++ /dev/null
@@ -1,92 +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.measure;
-
-import java.nio.ByteBuffer;
-import java.util.Collection;
-
-import org.apache.kylin.metadata.datatype.DataType;
-import org.apache.kylin.metadata.datatype.DataTypeSerializer;
-import org.apache.kylin.metadata.model.MeasureDesc;
-
-/**
- * @author yangli9
- *
- */
-@SuppressWarnings({ "rawtypes", "unchecked" })
-public class MeasureCodec {
-
- int nMeasures;
- DataTypeSerializer[] serializers;
-
- public MeasureCodec(Collection<MeasureDesc> measureDescs) {
- this((MeasureDesc[]) measureDescs.toArray(new MeasureDesc[measureDescs.size()]));
- }
-
- public MeasureCodec(MeasureDesc... measureDescs) {
- String[] dataTypes = new String[measureDescs.length];
- for (int i = 0; i < dataTypes.length; i++) {
- dataTypes[i] = measureDescs[i].getFunction().getReturnType();
- }
- init(dataTypes);
- }
-
- public MeasureCodec(DataType... dateTypes) {
- init(dateTypes);
- }
-
- public MeasureCodec(String... dataTypes) {
- init(dataTypes);
- }
-
- private void init(String[] dataTypes) {
- DataType[] typeInstances = new DataType[dataTypes.length];
- for (int i = 0; i < dataTypes.length; i++) {
- typeInstances[i] = DataType.getType(dataTypes[i]);
- }
- init(typeInstances);
- }
-
- private void init(DataType[] dataTypes) {
- nMeasures = dataTypes.length;
- serializers = new DataTypeSerializer[nMeasures];
-
- for (int i = 0; i < nMeasures; i++) {
- serializers[i] = DataTypeSerializer.create(dataTypes[i]);
- }
- }
-
- public DataTypeSerializer getSerializer(int idx) {
- return serializers[idx];
- }
-
- public void decode(ByteBuffer buf, Object[] result) {
- assert result.length == nMeasures;
- for (int i = 0; i < nMeasures; i++) {
- result[i] = serializers[i].deserialize(buf);
- }
- }
-
- public void encode(Object[] values, ByteBuffer out) {
- assert values.length == nMeasures;
- for (int i = 0; i < nMeasures; i++) {
- serializers[i].serialize(values[i], out);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/core-metadata/src/main/java/org/apache/kylin/measure/MeasureDecoder.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureDecoder.java b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureDecoder.java
new file mode 100644
index 0000000..63dd1e7
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureDecoder.java
@@ -0,0 +1,86 @@
+/*
+ * 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.measure;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.datatype.DataTypeSerializer;
+import org.apache.kylin.metadata.model.MeasureDesc;
+
+/**
+ * @author yangli9
+ *
+ */
+@SuppressWarnings({ "rawtypes" })
+public class MeasureDecoder {
+
+ int nMeasures;
+ DataTypeSerializer[] serializers;
+
+ public MeasureDecoder(Collection<MeasureDesc> measureDescs) {
+ this((MeasureDesc[]) measureDescs.toArray(new MeasureDesc[measureDescs.size()]));
+ }
+
+ public MeasureDecoder(MeasureDesc... measureDescs) {
+ String[] dataTypes = new String[measureDescs.length];
+ for (int i = 0; i < dataTypes.length; i++) {
+ dataTypes[i] = measureDescs[i].getFunction().getReturnType();
+ }
+ init(dataTypes);
+ }
+
+ public MeasureDecoder(DataType... dataTypes) {
+ init(dataTypes);
+ }
+
+ public MeasureDecoder(String... dataTypes) {
+ init(dataTypes);
+ }
+
+ private void init(String[] dataTypes) {
+ DataType[] typeInstances = new DataType[dataTypes.length];
+ for (int i = 0; i < dataTypes.length; i++) {
+ typeInstances[i] = DataType.getType(dataTypes[i]);
+ }
+ init(typeInstances);
+ }
+
+ private void init(DataType[] dataTypes) {
+ nMeasures = dataTypes.length;
+ serializers = new DataTypeSerializer[nMeasures];
+
+ for (int i = 0; i < nMeasures; i++) {
+ serializers[i] = DataTypeSerializer.create(dataTypes[i]);
+ }
+ }
+
+ public DataTypeSerializer getSerializer(int idx) {
+ return serializers[idx];
+ }
+
+ public void decode(ByteBuffer buf, Object[] result) {
+ assert result.length == nMeasures;
+ for (int i = 0; i < nMeasures; i++) {
+ result[i] = serializers[i].deserialize(buf);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
----------------------------------------------------------------------
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
index 4dea46b..740e896 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureType.java
@@ -107,6 +107,7 @@ abstract public class MeasureType<T> {
/** Whether or not Calcite rel-tree needs rewrite to do last around of aggregation */
abstract public boolean needRewrite();
+ /** Does the rewrite involves an extra field for the pre-calculated */
public boolean needRewriteField() {
return true;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
index cb4b1cb..78b272c 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/common/CuboidStatsUtil.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.cube.kv.RowConstants;
@@ -49,7 +50,7 @@ public class CuboidStatsUtil {
allCuboids.addAll(cuboidHLLMap.keySet());
Collections.sort(allCuboids);
- ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ ByteBuffer valueBuf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);
SequenceFile.Writer writer = SequenceFile.createWriter(conf, SequenceFile.Writer.file(seqFilePath), SequenceFile.Writer.keyClass(LongWritable.class), SequenceFile.Writer.valueClass(BytesWritable.class));
try {
// mapper overlap ratio at key -1
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
index a1eeb1b..e404f9a 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/BaseCuboidMapperBase.java
@@ -35,14 +35,13 @@ import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.kv.AbstractRowKeyEncoder;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
import org.apache.kylin.dimension.Dictionary;
import org.apache.kylin.engine.mr.KylinMapper;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureIngester;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
@@ -77,11 +76,10 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL
protected byte[][] keyBytesBuf;
protected BytesSplitter bytesSplitter;
protected AbstractRowKeyEncoder rowKeyEncoder;
- protected MeasureCodec measureCodec;
+ protected BufferedMeasureEncoder measureCodec;
private int errorRecordCounter;
protected Text outputKey = new Text();
protected Text outputValue = new Text();
- private ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
@Override
protected void setup(Context context) throws IOException {
@@ -110,7 +108,7 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL
bytesSplitter = new BytesSplitter(200, 16384);
rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, baseCuboid);
- measureCodec = new MeasureCodec(cubeDesc.getMeasures());
+ measureCodec = new BufferedMeasureEncoder(cubeDesc.getMeasures());
measures = new Object[cubeDesc.getMeasures().size()];
int colCount = cubeDesc.getRowkey().getRowKeyColumns().length;
@@ -153,14 +151,13 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL
return rowKeyEncoder.encode(keyBytesBuf);
}
- private void buildValue(SplittedBytes[] splitBuffers) {
+ private ByteBuffer buildValue(SplittedBytes[] splitBuffers) {
for (int i = 0; i < measures.length; i++) {
measures[i] = buildValueOf(i, splitBuffers);
}
- valueBuf.clear();
- measureCodec.encode(measures, valueBuf);
+ return measureCodec.encode(measures);
}
private Object buildValueOf(int idxOfMeasure, SplittedBytes[] splitBuffers) {
@@ -203,7 +200,7 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL
byte[] rowKey = buildKey(bytesSplitter.getSplitBuffers());
outputKey.set(rowKey, 0, rowKey.length);
- buildValue(bytesSplitter.getSplitBuffers());
+ ByteBuffer valueBuf = buildValue(bytesSplitter.getSplitBuffers());
outputValue.set(valueBuf.array(), 0, valueBuf.position());
context.write(outputKey, outputValue);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidReducer.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidReducer.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidReducer.java
index f263d99..6986344 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidReducer.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/CuboidReducer.java
@@ -25,13 +25,12 @@ import java.util.List;
import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.KylinReducer;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureAggregators;
-import org.apache.kylin.measure.MeasureCodec;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,7 +47,7 @@ public class CuboidReducer extends KylinReducer<Text, Text, Text, Text> {
private CubeDesc cubeDesc;
private List<MeasureDesc> measuresDescs;
- private MeasureCodec codec;
+ private BufferedMeasureEncoder codec;
private MeasureAggregators aggs;
private int counter;
@@ -57,7 +56,6 @@ public class CuboidReducer extends KylinReducer<Text, Text, Text, Text> {
private Object[] input;
private Object[] result;
- private ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
private Text outputValue = new Text();
@Override
@@ -73,7 +71,7 @@ public class CuboidReducer extends KylinReducer<Text, Text, Text, Text> {
cubeDesc = CubeManager.getInstance(config).getCube(cubeName).getDescriptor();
measuresDescs = cubeDesc.getMeasures();
- codec = new MeasureCodec(measuresDescs);
+ codec = new BufferedMeasureEncoder(measuresDescs);
aggs = new MeasureAggregators(measuresDescs);
input = new Object[measuresDescs.size()];
@@ -101,8 +99,7 @@ public class CuboidReducer extends KylinReducer<Text, Text, Text, Text> {
}
aggs.collectStates(result);
- valueBuf.clear();
- codec.encode(result, valueBuf);
+ ByteBuffer valueBuf = codec.encode(result);
outputValue.set(valueBuf.array(), 0, valueBuf.position());
context.write(key, outputValue);
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
index 2688919..3be5795 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/FactDistinctHiveColumnsMapper.java
@@ -24,6 +24,7 @@ import java.util.Collection;
import java.util.List;
import org.apache.hadoop.io.Text;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.Bytes;
@@ -159,7 +160,7 @@ public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMap
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
if (collectStatistics) {
- ByteBuffer hllBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ ByteBuffer hllBuf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);
// output each cuboid's hll to reducer, key is 0 - cuboidId
HyperLogLogPlusCounter hll;
for (int i = 0; i < cuboidIds.length; i++) {
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
index ec36242..673cfc0 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/InMemCuboidReducer.java
@@ -26,17 +26,14 @@ import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.CubeSegment;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.ByteArrayWritable;
import org.apache.kylin.engine.mr.KylinReducer;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureAggregators;
-import org.apache.kylin.measure.MeasureCodec;
import org.apache.kylin.metadata.model.MeasureDesc;
-import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,7 +43,7 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra
private static final Logger logger = LoggerFactory.getLogger(InMemCuboidReducer.class);
- private MeasureCodec codec;
+ private BufferedMeasureEncoder codec;
private MeasureAggregators aggs;
private int counter;
@@ -55,7 +52,6 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra
private Text outputKey;
private Text outputValue;
- private ByteBuffer valueBuf;
@Override
protected void setup(Context context) throws IOException {
@@ -63,22 +59,17 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra
KylinConfig config = AbstractHadoopJob.loadKylinPropsAndMetadata();
String cubeName = context.getConfiguration().get(BatchConstants.CFG_CUBE_NAME).toUpperCase();
- String segmentName = context.getConfiguration().get(BatchConstants.CFG_CUBE_SEGMENT_NAME);
- boolean isMerge = Boolean.parseBoolean(context.getConfiguration().get(BatchConstants.CFG_IS_MERGE));
-
CubeInstance cube = CubeManager.getInstance(config).getCube(cubeName);
CubeDesc cubeDesc = cube.getDescriptor();
- CubeSegment cubeSeg = cube.getSegment(segmentName, SegmentStatusEnum.NEW);
List<MeasureDesc> measuresDescs = cubeDesc.getMeasures();
- codec = new MeasureCodec(measuresDescs);
+ codec = new BufferedMeasureEncoder(measuresDescs);
aggs = new MeasureAggregators(measuresDescs);
input = new Object[measuresDescs.size()];
result = new Object[measuresDescs.size()];
outputKey = new Text();
outputValue = new Text();
- valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
}
@Override
@@ -96,8 +87,7 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra
outputKey.set(key.array(), key.offset(), key.length());
// output value
- valueBuf.clear();
- codec.encode(result, valueBuf);
+ ByteBuffer valueBuf = codec.encode(result);
outputValue.set(valueBuf.array(), 0, valueBuf.position());
context.write(outputKey, outputValue);
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
index ee65132..de8b6d4 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/KVGTRecordWriter.java
@@ -18,19 +18,21 @@
package org.apache.kylin.engine.mr.steps;
+import java.io.IOException;
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.inmemcubing.ICuboidWriter;
import org.apache.kylin.cube.kv.AbstractRowKeyEncoder;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.ByteArrayWritable;
import org.apache.kylin.gridtable.GTRecord;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
/**
*/
@@ -45,8 +47,8 @@ public abstract class KVGTRecordWriter implements ICuboidWriter {
private int dimensions;
private int measureCount;
private byte[] keyBuf;
- private int[] measureColumnsIndex;
- private ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ private ImmutableBitSet measureColumns;
+ private ByteBuffer valueBuf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);
private ByteArrayWritable outputKey = new ByteArrayWritable();
private ByteArrayWritable outputValue = new ByteArrayWritable();
private long cuboidRowCount = 0;
@@ -77,7 +79,12 @@ public abstract class KVGTRecordWriter implements ICuboidWriter {
//output measures
valueBuf.clear();
- record.exportColumns(measureColumnsIndex, valueBuf);
+ try {
+ record.exportColumns(measureColumns, valueBuf);
+ } catch (BufferOverflowException boe) {
+ valueBuf = ByteBuffer.allocate((int) (record.sizeOf(measureColumns) * 1.5));
+ record.exportColumns(measureColumns, valueBuf);
+ }
outputKey.set(keyBuf, 0, keyBuf.length);
outputValue.set(valueBuf.array(), 0, valueBuf.position());
@@ -91,9 +98,6 @@ public abstract class KVGTRecordWriter implements ICuboidWriter {
keyBuf = rowKeyEncoder.createBuf();
dimensions = Long.bitCount(cuboidId);
- measureColumnsIndex = new int[measureCount];
- for (int i = 0; i < measureCount; i++) {
- measureColumnsIndex[i] = dimensions + i;
- }
+ measureColumns = new ImmutableBitSet(dimensions, dimensions + measureCount);
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
index 6766b31..72de7dc 100644
--- a/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
+++ b/engine-mr/src/main/java/org/apache/kylin/engine/mr/steps/MergeCuboidMapper.java
@@ -47,7 +47,7 @@ import org.apache.kylin.dimension.Dictionary;
import org.apache.kylin.engine.mr.KylinMapper;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureIngester;
import org.apache.kylin.measure.MeasureType;
import org.apache.kylin.metadata.model.MeasureDesc;
@@ -85,9 +85,8 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
private Map<TblColRef, Dictionary<String>> oldDicts;
private Map<TblColRef, Dictionary<String>> newDicts;
private List<MeasureDesc> measureDescs;
- private MeasureCodec codec;
+ private BufferedMeasureEncoder codec;
private Object[] measureObjs;
- private ByteBuffer valueBuf;
private Text outputValue;
@Override
@@ -116,9 +115,8 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment);
measureDescs = cubeDesc.getMeasures();
- codec = new MeasureCodec(measureDescs);
+ codec = new BufferedMeasureEncoder(measureDescs);
measureObjs = new Object[measureDescs.size()];
- valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
outputValue = new Text();
dictMeasures = Lists.newArrayList();
@@ -233,8 +231,7 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> {
MeasureIngester ingester = pair.getSecond();
measureObjs[i] = ingester.reEncodeDictionary(measureObjs[i], measureDescs.get(i), oldDicts, newDicts);
}
- valueBuf.clear();
- codec.encode(measureObjs, valueBuf);
+ ByteBuffer valueBuf = codec.encode(measureObjs);
outputValue.set(valueBuf.array(), 0, valueBuf.position());
value = outputValue;
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
----------------------------------------------------------------------
diff --git a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
index aa5e77e..b64afd3 100644
--- a/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
+++ b/engine-mr/src/test/java/org/apache/kylin/engine/mr/steps/CubeReducerTest.java
@@ -18,8 +18,7 @@
package org.apache.kylin.engine.mr.steps;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
import java.io.File;
import java.lang.reflect.Field;
@@ -34,11 +33,10 @@ import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
import org.apache.hadoop.mrunit.types.Pair;
import org.apache.kylin.common.util.LocalFileMetadataTestCase;
import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.common.BatchConstants;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureAggregator;
-import org.apache.kylin.measure.MeasureCodec;
import org.apache.kylin.measure.MeasureIngester;
import org.apache.kylin.measure.MeasureType;
import org.apache.kylin.metadata.datatype.LongMutable;
@@ -50,12 +48,11 @@ import org.junit.Test;
/**
*/
+@SuppressWarnings("rawtypes")
public class CubeReducerTest extends LocalFileMetadataTestCase {
ReduceDriver<Text, Text, Text, Text> reduceDriver;
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
-
@Before
public void setUp() throws Exception {
createTestMetadata();
@@ -80,7 +77,7 @@ public class CubeReducerTest extends LocalFileMetadataTestCase {
reduceDriver.getConfiguration().set(BatchConstants.CFG_CUBE_NAME, "test_kylin_cube_with_slr_ready");
CubeDesc cubeDesc = CubeManager.getInstance(getTestConfig()).getCube("test_kylin_cube_with_slr_ready").getDescriptor();
- MeasureCodec codec = new MeasureCodec(cubeDesc.getMeasures());
+ BufferedMeasureEncoder codec = new BufferedMeasureEncoder(cubeDesc.getMeasures());
Text key1 = new Text("72010ustech");
List<Text> values1 = new ArrayList<Text>();
@@ -127,7 +124,7 @@ public class CubeReducerTest extends LocalFileMetadataTestCase {
MeasureType origMeasureType = functionDesc.getMeasureType();
field.set(functionDesc, new MockUpMeasureType(origMeasureType));
- MeasureCodec codec = new MeasureCodec(cubeDesc.getMeasures());
+ BufferedMeasureEncoder codec = new BufferedMeasureEncoder(cubeDesc.getMeasures());
Text key1 = new Text("72010ustech");
List<Text> values1 = new ArrayList<Text>();
@@ -161,11 +158,10 @@ public class CubeReducerTest extends LocalFileMetadataTestCase {
assertTrue(result.contains(p3));
}
- private Text newValueText(MeasureCodec codec, String sum, String min, String max, int count, int item_count) {
+ private Text newValueText(BufferedMeasureEncoder codec, String sum, String min, String max, int count, int item_count) {
Object[] values = new Object[] { new BigDecimal(sum), new BigDecimal(min), new BigDecimal(max), new LongMutable(count), new LongMutable(item_count) };
- buf.clear();
- codec.encode(values, buf);
+ ByteBuffer buf = codec.encode(values);
Text t = new Text();
t.set(buf.array(), 0, buf.position());
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
----------------------------------------------------------------------
diff --git a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
index 2b4e497..796fb9e 100644
--- a/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
+++ b/engine-spark/src/main/java/org/apache/kylin/engine/spark/SparkCubing.java
@@ -67,7 +67,6 @@ import org.apache.kylin.cube.cuboid.CuboidScheduler;
import org.apache.kylin.cube.inmemcubing.AbstractInMemCubeBuilder;
import org.apache.kylin.cube.inmemcubing.DoggedCubeBuilder;
import org.apache.kylin.cube.kv.CubeDimEncMap;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.CubeJoinedFlatTableDesc;
import org.apache.kylin.cube.model.DimensionDesc;
@@ -81,8 +80,8 @@ import org.apache.kylin.engine.mr.common.CubeStatsReader;
import org.apache.kylin.engine.spark.cube.BufferedCuboidWriter;
import org.apache.kylin.engine.spark.cube.DefaultTupleConverter;
import org.apache.kylin.engine.spark.util.IteratorUtils;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureAggregators;
-import org.apache.kylin.measure.MeasureCodec;
import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
@@ -110,8 +109,6 @@ import org.reflections.Reflections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.Tuple2;
-
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
@@ -123,6 +120,8 @@ import com.google.common.hash.Hasher;
import com.google.common.hash.Hashing;
import com.google.common.primitives.UnsignedBytes;
+import scala.Tuple2;
+
/**
*/
public class SparkCubing extends AbstractApplication {
@@ -422,8 +421,7 @@ public class SparkCubing extends AbstractApplication {
@Override
public Iterable<Tuple2<byte[], byte[]>> call(final Iterator<Tuple2<byte[], byte[]>> tuple2Iterator) throws Exception {
return new Iterable<Tuple2<byte[], byte[]>>() {
- final ByteBuffer buffer = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
- final MeasureCodec codec = new MeasureCodec(dataTypes);
+ final BufferedMeasureEncoder codec = new BufferedMeasureEncoder(dataTypes);
final Object[] input = new Object[measureSize];
final Object[] result = new Object[measureSize];
@@ -442,8 +440,7 @@ public class SparkCubing extends AbstractApplication {
aggs.aggregate(input);
}
aggs.collectStates(result);
- buffer.clear();
- codec.encode(result, buffer);
+ ByteBuffer buffer = codec.encode(result);
byte[] bytes = new byte[buffer.position()];
System.arraycopy(buffer.array(), buffer.arrayOffset(), bytes, 0, buffer.position());
return bytes;
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
----------------------------------------------------------------------
diff --git a/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java b/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
index 8da1519..2532679 100644
--- a/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
+++ b/engine-spark/src/main/java/org/apache/kylin/engine/spark/cube/DefaultTupleConverter.java
@@ -17,16 +17,19 @@
*/
package org.apache.kylin.engine.spark.cube;
+import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.util.Map;
import org.apache.kylin.common.util.ByteArray;
+import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.kv.RowKeyEncoder;
import org.apache.kylin.cube.kv.RowKeyEncoderProvider;
import org.apache.kylin.gridtable.GTRecord;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.metadata.model.TblColRef;
import scala.Tuple2;
@@ -36,7 +39,6 @@ import scala.Tuple2;
public final class DefaultTupleConverter implements TupleConverter {
private final static ThreadLocal<ByteBuffer> valueBuf = new ThreadLocal<>();
- private final static ThreadLocal<int[]> measureColumnsIndex = new ThreadLocal<>();
private final CubeSegment segment;
private final int measureCount;
private final Map<TblColRef, Integer> columnLengthMap;
@@ -52,16 +54,13 @@ public final class DefaultTupleConverter implements TupleConverter {
private ByteBuffer getValueBuf() {
if (valueBuf.get() == null) {
- valueBuf.set(ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE));
+ valueBuf.set(ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE));
}
return valueBuf.get();
}
-
- private int[] getMeasureColumnsIndex() {
- if (measureColumnsIndex.get() == null) {
- measureColumnsIndex.set(new int[measureCount]);
- }
- return measureColumnsIndex.get();
+
+ private void setValueBuf(ByteBuffer buf) {
+ valueBuf.set(buf);
}
@Override
@@ -70,10 +69,7 @@ public final class DefaultTupleConverter implements TupleConverter {
RowKeyEncoder rowkeyEncoder = rowKeyEncoderProvider.getRowkeyEncoder(cuboid);
final int dimensions = Long.bitCount(cuboidId);
- int[] measureColumnsIndex = getMeasureColumnsIndex();
- for (int i = 0; i < measureCount; i++) {
- measureColumnsIndex[i] = dimensions + i;
- }
+ final ImmutableBitSet measureColumns = new ImmutableBitSet(dimensions, dimensions + measureCount);
int offSet = 0;
for (int x = 0; x < dimensions; x++) {
@@ -87,7 +83,13 @@ public final class DefaultTupleConverter implements TupleConverter {
ByteBuffer valueBuf = getValueBuf();
valueBuf.clear();
- record.exportColumns(measureColumnsIndex, valueBuf);
+ try {
+ record.exportColumns(measureColumns, valueBuf);
+ } catch (BufferOverflowException boe) {
+ valueBuf = ByteBuffer.allocate((int) (record.sizeOf(measureColumns) * 1.5));
+ record.exportColumns(measureColumns, valueBuf);
+ setValueBuf(valueBuf);
+ }
byte[] value = new byte[valueBuf.position()];
System.arraycopy(valueBuf.array(), 0, value, 0, valueBuf.position());
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
index 68b6ae4..e06fb68 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityMapper.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.cube.kv.RowConstants;
@@ -97,7 +98,7 @@ public class ColumnCardinalityMapper<T> extends KylinMapper<T, Object, IntWritab
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
Iterator<Integer> it = hllcMap.keySet().iterator();
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);
while (it.hasNext()) {
int key = it.next();
HyperLogLogPlusCounter hllc = hllcMap.get(key);
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducer.java
----------------------------------------------------------------------
diff --git a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducer.java b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducer.java
index 19f5759..6596917 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducer.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducer.java
@@ -30,6 +30,7 @@ import java.util.Map;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.engine.mr.KylinReducer;
@@ -79,7 +80,7 @@ public class ColumnCardinalityReducer extends KylinReducer<IntWritable, BytesWri
while (it.hasNext()) {
int key = it.next();
HyperLogLogPlusCounter hllc = hllcMap.get(key);
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);
buf.clear();
hllc.writeRegisters(buf);
buf.flip();
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/source-hive/src/test/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducerTest.java
----------------------------------------------------------------------
diff --git a/source-hive/src/test/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducerTest.java b/source-hive/src/test/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducerTest.java
index 57721d6..b9f532d 100644
--- a/source-hive/src/test/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducerTest.java
+++ b/source-hive/src/test/java/org/apache/kylin/source/hive/cardinality/ColumnCardinalityReducerTest.java
@@ -33,6 +33,7 @@ import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
import org.apache.hadoop.mrunit.types.Pair;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.cube.kv.RowConstants;
@@ -65,7 +66,7 @@ public class ColumnCardinalityReducerTest {
i++;
hllc.add(Bytes.toBytes(temp));
}
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);
buf.clear();
hllc.writeRegisters(buf);
buf.flip();
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregators.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregators.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregators.java
index 9e9dd6d..941c10f 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregators.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/ObserverAggregators.java
@@ -28,10 +28,9 @@ import org.apache.hadoop.hbase.Cell;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.BytesSerializer;
import org.apache.kylin.common.util.BytesUtil;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.HBaseColumnDesc;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.measure.MeasureAggregator;
-import org.apache.kylin.measure.MeasureCodec;
import org.apache.kylin.measure.MeasureType;
import org.apache.kylin.measure.MeasureTypeFactory;
import org.apache.kylin.metadata.datatype.DataType;
@@ -228,9 +227,7 @@ public class ObserverAggregators {
for (int j = 0; j < col.nMeasures; j++)
col.measureValues[j] = aggrs[i++].getState();
- col.measureBuf.clear();
- col.measureCodec.encode(col.measureValues, col.measureBuf);
- hColValues[ci] = col.measureBuf;
+ hColValues[ci] = col.measureCodec.encode(col.measureValues);
}
return hColValues;
}
@@ -244,9 +241,8 @@ public class ObserverAggregators {
final String[] dataTypes;
final int nMeasures;
- final MeasureCodec measureCodec;
+ final BufferedMeasureEncoder measureCodec;
final Object[] measureValues;
- final ByteBuffer measureBuf;
public HCol(byte[] bFamily, byte[] bQualifier, String[] funcNames, String[] dataTypes) {
this.family = bFamily;
@@ -256,9 +252,8 @@ public class ObserverAggregators {
this.nMeasures = funcNames.length;
assert funcNames.length == dataTypes.length;
- this.measureCodec = new MeasureCodec(dataTypes);
+ this.measureCodec = new BufferedMeasureEncoder(dataTypes);
this.measureValues = new Object[nMeasures];
- this.measureBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
}
@Override
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
index 510e735..3ccf7cf 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/cube/v2/coprocessor/endpoint/CubeVisitService.java
@@ -22,6 +22,7 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
+import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
@@ -48,6 +49,7 @@ 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.measure.BufferedMeasureEncoder;
import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.IRealizationConstants;
@@ -279,9 +281,9 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
behavior.ordinal() >= CoprocessorBehavior.SCAN_FILTER.ordinal(), //
behavior.ordinal() >= CoprocessorBehavior.SCAN_FILTER_AGGR.ordinal());
- ByteBuffer buffer = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream(RowConstants.ROWVALUE_BUFFER_SIZE);//ByteArrayOutputStream will auto grow
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);//ByteArrayOutputStream will auto grow
int finalRowCount = 0;
for (GTRecord oneRecord : finalScanner) {
@@ -298,10 +300,14 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement
}
buffer.clear();
- oneRecord.exportColumns(scanReq.getColumns(), buffer);
- buffer.flip();
+ try {
+ oneRecord.exportColumns(scanReq.getColumns(), buffer);
+ } catch (BufferOverflowException boe) {
+ buffer = ByteBuffer.allocate((int) (oneRecord.sizeOf(scanReq.getColumns()) * 1.5));
+ oneRecord.exportColumns(scanReq.getColumns(), buffer);
+ }
- outputStream.write(buffer.array(), buffer.arrayOffset() - buffer.position(), buffer.remaining());
+ outputStream.write(buffer.array(), 0, buffer.position());
finalRowCount++;
}
finalScanner.close();
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper.java
index 8205ff7..7523249 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper.java
@@ -33,7 +33,7 @@ import org.apache.kylin.cube.model.HBaseColumnFamilyDesc;
import org.apache.kylin.engine.mr.KylinMapper;
import org.apache.kylin.engine.mr.common.AbstractHadoopJob;
import org.apache.kylin.engine.mr.common.BatchConstants;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.MeasureDecoder;
import com.google.common.collect.Lists;
@@ -48,7 +48,7 @@ public class CubeHFileMapper extends KylinMapper<Text, Text, ImmutableBytesWrita
String cubeName;
CubeDesc cubeDesc;
- MeasureCodec inputCodec;
+ MeasureDecoder inputCodec;
Object[] inputMeasures;
List<KeyValueCreator> keyValueCreators;
@@ -62,7 +62,7 @@ public class CubeHFileMapper extends KylinMapper<Text, Text, ImmutableBytesWrita
CubeManager cubeMgr = CubeManager.getInstance(config);
cubeDesc = cubeMgr.getCube(cubeName).getDescriptor();
- inputCodec = new MeasureCodec(cubeDesc.getMeasures());
+ inputCodec = new MeasureDecoder(cubeDesc.getMeasures());
inputMeasures = new Object[cubeDesc.getMeasures().size()];
keyValueCreators = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/KeyValueCreator.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/KeyValueCreator.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/KeyValueCreator.java
index c6a233b..490031e 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/KeyValueCreator.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/KeyValueCreator.java
@@ -24,10 +24,9 @@ import java.util.List;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.io.Text;
import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.HBaseColumnDesc;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.metadata.model.MeasureDesc;
/**
@@ -41,9 +40,8 @@ public class KeyValueCreator {
int[] refIndex;
MeasureDesc[] refMeasures;
- MeasureCodec codec;
+ BufferedMeasureEncoder codec;
Object[] colValues;
- ByteBuffer valueBuf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
public boolean isFullCopy;
@@ -56,7 +54,7 @@ public class KeyValueCreator {
refIndex = colDesc.getMeasureIndex();
refMeasures = colDesc.getMeasures();
- codec = new MeasureCodec(refMeasures);
+ codec = new BufferedMeasureEncoder(refMeasures);
colValues = new Object[refMeasures.length];
isFullCopy = true;
@@ -76,8 +74,7 @@ public class KeyValueCreator {
colValues[i] = measureValues[refIndex[i]];
}
- valueBuf.clear();
- codec.encode(colValues, valueBuf);
+ ByteBuffer valueBuf = codec.encode(colValues);
return create(keyBytes, keyOffset, keyLength, valueBuf.array(), 0, valueBuf.position());
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
index 7d05dc9..e1e8f8c 100644
--- a/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
+++ b/storage-hbase/src/main/java/org/apache/kylin/storage/hbase/steps/RowValueDecoder.java
@@ -25,7 +25,7 @@ import java.util.Collection;
import org.apache.hadoop.hbase.client.Result;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.cube.model.HBaseColumnDesc;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.MeasureDecoder;
import org.apache.kylin.metadata.datatype.DoubleMutable;
import org.apache.kylin.metadata.datatype.LongMutable;
import org.apache.kylin.metadata.model.FunctionDesc;
@@ -44,7 +44,7 @@ public class RowValueDecoder implements Cloneable {
private final byte[] hbaseColumnFamily;
private final byte[] hbaseColumnQualifier;
- private final MeasureCodec codec;
+ private final MeasureDecoder codec;
private final BitSet projectionIndex;
private final MeasureDesc[] measures;
private final Object[] values;
@@ -55,7 +55,7 @@ public class RowValueDecoder implements Cloneable {
this.hbaseColumnQualifier = Bytes.toBytes(hbaseColumn.getQualifier());
this.projectionIndex = new BitSet();
this.measures = hbaseColumn.getMeasures();
- this.codec = new MeasureCodec(measures);
+ this.codec = new MeasureDecoder(measures);
this.values = new Object[measures.length];
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
index cd4e33d..3efcf92 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/cube/v1/coprocessor/observer/AggregateRegionObserverTest.java
@@ -18,8 +18,7 @@
package org.apache.kylin.storage.hbase.cube.v1.coprocessor.observer;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
import java.io.IOException;
import java.math.BigDecimal;
@@ -36,7 +35,6 @@ import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.Type;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.kylin.common.util.Bytes;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.metadata.datatype.LongMutable;
import org.apache.kylin.metadata.model.ColumnDesc;
import org.apache.kylin.metadata.model.TableDesc;
@@ -55,7 +53,6 @@ import com.google.common.collect.Lists;
* @author yangli9
*/
public class AggregateRegionObserverTest {
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
byte[] mask = new byte[] { (byte) 0xff, (byte) 0xff, 0, 0 };
byte[] k1 = new byte[] { 0x01, 0x01, 0, 0x01 };
@@ -94,8 +91,7 @@ public class AggregateRegionObserverTest {
Object[] values = number == Integer.MIN_VALUE ? //
new Object[] { new BigDecimal(decimal) } //
: new Object[] { new BigDecimal(decimal), new LongMutable(number) };
- buf.clear();
- col.measureCodec.encode(values, buf);
+ ByteBuffer buf = col.measureCodec.encode(values);
Cell keyValue = new KeyValue(key, 0, key.length, //
col.family, 0, col.family.length, //
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
index b715498..dbf39e7 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/CubeHFileMapper2Test.java
@@ -35,7 +35,7 @@ import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.engine.mr.HadoopUtil;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.MeasureDecoder;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -49,8 +49,7 @@ public class CubeHFileMapper2Test extends LocalFileMetadataTestCase {
String cubeName = "test_kylin_cube_with_slr_ready";
- MeasureCodec codec;
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
+ MeasureDecoder codec;
Object[] outKV = new Object[2];
@Before
@@ -60,7 +59,7 @@ public class CubeHFileMapper2Test extends LocalFileMetadataTestCase {
FileUtils.deleteDirectory(new File("../job/meta"));
FileUtils.copyDirectory(new File(getTestConfig().getMetadataUrl()), new File("../job/meta"));
CubeDesc desc = CubeManager.getInstance(getTestConfig()).getCube(cubeName).getDescriptor();
- codec = new MeasureCodec(desc.getMeasures());
+ codec = new MeasureDecoder(desc.getMeasures());
}
@After
http://git-wip-us.apache.org/repos/asf/kylin/blob/d56abdea/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RowValueDecoderTest.java
----------------------------------------------------------------------
diff --git a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RowValueDecoderTest.java b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RowValueDecoderTest.java
index 488ff59..6475bad 100644
--- a/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RowValueDecoderTest.java
+++ b/storage-hbase/src/test/java/org/apache/kylin/storage/hbase/steps/RowValueDecoderTest.java
@@ -18,7 +18,7 @@
package org.apache.kylin.storage.hbase.steps;
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
@@ -27,10 +27,9 @@ import java.util.Arrays;
import org.apache.hadoop.io.LongWritable;
import org.apache.kylin.common.util.LocalFileMetadataTestCase;
import org.apache.kylin.cube.CubeManager;
-import org.apache.kylin.cube.kv.RowConstants;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.HBaseColumnDesc;
-import org.apache.kylin.measure.MeasureCodec;
+import org.apache.kylin.measure.BufferedMeasureEncoder;
import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.datatype.LongMutable;
import org.apache.kylin.metadata.model.FunctionDesc;
@@ -57,14 +56,13 @@ public class RowValueDecoderTest extends LocalFileMetadataTestCase {
CubeDesc cubeDesc = CubeManager.getInstance(getTestConfig()).getCube("test_kylin_cube_with_slr_ready").getDescriptor();
HBaseColumnDesc hbaseCol = cubeDesc.getHbaseMapping().getColumnFamily()[0].getColumns()[0];
- MeasureCodec codec = new MeasureCodec(hbaseCol.getMeasures());
+ BufferedMeasureEncoder codec = new BufferedMeasureEncoder(hbaseCol.getMeasures());
BigDecimal sum = new BigDecimal("333.1234567");
BigDecimal min = new BigDecimal("333.1111111");
BigDecimal max = new BigDecimal("333.1999999");
LongMutable count = new LongMutable(2);
LongMutable item_count = new LongMutable(100);
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
- codec.encode(new Object[] { sum, min, max, count, item_count }, buf);
+ ByteBuffer buf = codec.encode(new Object[] { sum, min, max, count, item_count });
buf.flip();
byte[] valueBytes = new byte[buf.limit()];
@@ -88,14 +86,13 @@ public class RowValueDecoderTest extends LocalFileMetadataTestCase {
CubeDesc cubeDesc = CubeManager.getInstance(getTestConfig()).getCube("test_kylin_cube_with_slr_ready").getDescriptor();
HBaseColumnDesc hbaseCol = cubeDesc.getHbaseMapping().getColumnFamily()[0].getColumns()[0];
- MeasureCodec codec = new MeasureCodec(hbaseCol.getMeasures());
+ BufferedMeasureEncoder codec = new BufferedMeasureEncoder(hbaseCol.getMeasures());
BigDecimal sum = new BigDecimal("11111111111111111111333.1234567");
BigDecimal min = new BigDecimal("333.1111111");
BigDecimal max = new BigDecimal("333.1999999");
LongWritable count = new LongWritable(2);
LongMutable item_count = new LongMutable(100);
- ByteBuffer buf = ByteBuffer.allocate(RowConstants.ROWVALUE_BUFFER_SIZE);
- codec.encode(new Object[] { sum, min, max, count, item_count }, buf);
+ codec.encode(new Object[] { sum, min, max, count, item_count });
}
}