refactor: rename MeasureDecoder to MeasureCodec
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/fa5c9cb7 Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/fa5c9cb7 Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/fa5c9cb7 Branch: refs/heads/1.5.x-HBase1.x Commit: fa5c9cb7851afff028f5a2d39d12fbbb32bd40e8 Parents: 4c9b67f Author: Hongbin Ma <mahong...@apache.org> Authored: Thu Sep 22 11:22:32 2016 +0800 Committer: Hongbin Ma <mahong...@apache.org> Committed: Thu Sep 22 18:50:35 2016 +0800 ---------------------------------------------------------------------- .../kylin/gridtable/GTAggregateScanner.java | 8 +- .../metadata/measure/MeasureCodecTest.java | 4 +- .../kylin/measure/BufferedMeasureCodec.java | 107 +++++++++++++++++++ .../kylin/measure/BufferedMeasureEncoder.java | 107 ------------------- .../org/apache/kylin/measure/MeasureCodec.java | 101 +++++++++++++++++ .../apache/kylin/measure/MeasureDecoder.java | 97 ----------------- .../kylin/engine/mr/common/CuboidStatsUtil.java | 4 +- .../engine/mr/steps/BaseCuboidMapperBase.java | 6 +- .../kylin/engine/mr/steps/CuboidReducer.java | 6 +- .../mr/steps/FactDistinctHiveColumnsMapper.java | 4 +- .../engine/mr/steps/InMemCuboidReducer.java | 6 +- .../kylin/engine/mr/steps/KVGTRecordWriter.java | 4 +- .../engine/mr/steps/MergeCuboidMapper.java | 6 +- .../kylin/engine/mr/steps/CubeReducerTest.java | 8 +- .../apache/kylin/engine/spark/SparkCubing.java | 4 +- .../spark/cube/DefaultTupleConverter.java | 4 +- .../cardinality/ColumnCardinalityMapper.java | 4 +- .../cardinality/ColumnCardinalityReducer.java | 4 +- .../ColumnCardinalityReducerTest.java | 4 +- .../observer/ObserverAggregators.java | 6 +- .../coprocessor/endpoint/CubeVisitService.java | 6 +- .../storage/hbase/steps/CubeHFileMapper.java | 6 +- .../storage/hbase/steps/KeyValueCreator.java | 6 +- .../storage/hbase/steps/RowValueDecoder.java | 6 +- .../hbase/steps/CubeHFileMapper2Test.java | 6 +- .../hbase/steps/RowValueDecoderTest.java | 6 +- 26 files changed, 267 insertions(+), 263 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 db38484..9bbcd10 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 @@ -40,7 +40,7 @@ import org.apache.kylin.common.util.ImmutableBitSet; import org.apache.kylin.common.util.MemoryBudgetController; import org.apache.kylin.common.util.Pair; import org.apache.kylin.common.util.MemoryBudgetController.MemoryWaterLevel; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureAggregator; import org.apache.kylin.measure.MeasureAggregators; import org.apache.kylin.metadata.datatype.DataType; @@ -178,7 +178,7 @@ public class GTAggregateScanner implements IGTScanner { final List<Dump> dumps; final int keyLength; final boolean[] compareMask; - final BufferedMeasureEncoder measureCodec; + final BufferedMeasureCodec measureCodec; final Comparator<byte[]> bytesComparator = new Comparator<byte[]>() { @Override @@ -212,13 +212,13 @@ public class GTAggregateScanner implements IGTScanner { measureCodec = createMeasureCodec(); } - private BufferedMeasureEncoder createMeasureCodec() { + private BufferedMeasureCodec createMeasureCodec() { DataType[] types = new DataType[metrics.trueBitCount()]; for (int i = 0; i < types.length; i++) { types[i] = info.getColumnType(metrics.trueBitAt(i)); } - BufferedMeasureEncoder result = new BufferedMeasureEncoder(types); + BufferedMeasureCodec result = new BufferedMeasureCodec(types); result.setBufferSize(info.getMaxColumnLength(metrics)); return result; } http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 1a7e4b3..64c8970 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 @@ -24,7 +24,7 @@ import java.math.BigDecimal; import java.nio.ByteBuffer; import org.apache.kylin.common.util.LocalFileMetadataTestCase; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.bitmap.BitmapCounter; import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter; import org.apache.kylin.metadata.datatype.DoubleMutable; @@ -52,7 +52,7 @@ public class MeasureCodecTest extends LocalFileMetadataTestCase { @Test public void basicTest() { MeasureDesc descs[] = new MeasureDesc[] { measure("double"), measure("long"), measure("decimal"), measure("HLLC16"), measure("bitmap") }; - BufferedMeasureEncoder codec = new BufferedMeasureEncoder(descs); + BufferedMeasureCodec codec = new BufferedMeasureCodec(descs); DoubleMutable d = new DoubleMutable(1.0); LongMutable l = new LongMutable(2); http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java ---------------------------------------------------------------------- diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java b/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.java new file mode 100644 index 0000000..8a5481c --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureCodec.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 BufferedMeasureCodec { + public static final int DEFAULT_BUFFER_SIZE = 1024 * 1024; // 1 MB + public static final int MAX_BUFFER_SIZE = 1 * 1024 * DEFAULT_BUFFER_SIZE; // 1 GB + + final private MeasureCodec codec; + + private ByteBuffer buf; + final private int[] measureSizes; + + public BufferedMeasureCodec(Collection<MeasureDesc> measureDescs) { + this.codec = new MeasureCodec(measureDescs); + this.measureSizes = new int[codec.getMeasuresCount()]; + } + + public BufferedMeasureCodec(MeasureDesc... measureDescs) { + this.codec = new MeasureCodec(measureDescs); + this.measureSizes = new int[codec.getMeasuresCount()]; + } + + public BufferedMeasureCodec(DataType... dataTypes) { + this.codec = new MeasureCodec(dataTypes); + this.measureSizes = new int[codec.getMeasuresCount()]; + } + + public BufferedMeasureCodec(String... dataTypes) { + this.codec = new MeasureCodec(dataTypes); + this.measureSizes = new int[codec.getMeasuresCount()]; + } + + /** 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.getMeasuresCount(); + + while (true) { + try { + buf.clear(); + for (int i = 0, pos = 0; i < codec.getMeasuresCount(); i++) { + codec.encode(i, 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/fa5c9cb7/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 deleted file mode 100644 index 2b14715..0000000 --- a/core-metadata/src/main/java/org/apache/kylin/measure/BufferedMeasureEncoder.java +++ /dev/null @@ -1,107 +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.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 = 1 * 1024 * DEFAULT_BUFFER_SIZE; // 1 GB - - 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/fa5c9cb7/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 new file mode 100644 index 0000000..edaf806 --- /dev/null +++ b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureCodec.java @@ -0,0 +1,101 @@ +/* + * 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 MeasureCodec { + + private int nMeasures; + private 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... dataTypes) { + init(dataTypes); + } + + 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 void encode(int idx, Object o, ByteBuffer buf) { + serializers[idx].serialize(o, buf); + } + + public int getMeasuresCount() { + return nMeasures; + } + + public int[] getPeekLength(ByteBuffer buf) { + int[] length = new int[nMeasures]; + int offset = 0; + for (int i = 0; i < nMeasures; i++) { + length[i] = serializers[i].peekLength(buf); + offset += length[i]; + buf.position(offset); + } + return length; + } + + 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/fa5c9cb7/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 deleted file mode 100644 index 9656020..0000000 --- a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureDecoder.java +++ /dev/null @@ -1,97 +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" }) -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 int[] getPeekLength(ByteBuffer buf) { - int[] length = new int[nMeasures]; - int offset = 0; - for (int i = 0; i < nMeasures; i++) { - length[i] = serializers[i].peekLength(buf); - offset += length[i]; - buf.position(offset); - } - return length; - } - - 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/fa5c9cb7/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 87c3211..d5b4b0d 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 @@ -32,7 +32,7 @@ import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.SequenceFile; import org.apache.kylin.common.util.Bytes; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter; public class CuboidStatsUtil { @@ -50,7 +50,7 @@ public class CuboidStatsUtil { allCuboids.addAll(cuboidHLLMap.keySet()); Collections.sort(allCuboids); - ByteBuffer valueBuf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE); + ByteBuffer valueBuf = ByteBuffer.allocate(BufferedMeasureCodec.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/fa5c9cb7/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 588b087..4f0d3fd 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 @@ -42,7 +42,7 @@ import org.apache.kylin.engine.EngineFactory; 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.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureIngester; import org.apache.kylin.metadata.model.FunctionDesc; import org.apache.kylin.metadata.model.MeasureDesc; @@ -76,7 +76,7 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL protected byte[][] keyBytesBuf; protected BytesSplitter bytesSplitter; protected AbstractRowKeyEncoder rowKeyEncoder; - protected BufferedMeasureEncoder measureCodec; + protected BufferedMeasureCodec measureCodec; private int errorRecordCounter; protected Text outputKey = new Text(); protected Text outputValue = new Text(); @@ -108,7 +108,7 @@ public class BaseCuboidMapperBase<KEYIN, VALUEIN> extends KylinMapper<KEYIN, VAL bytesSplitter = new BytesSplitter(200, 16384); rowKeyEncoder = AbstractRowKeyEncoder.createInstance(cubeSegment, baseCuboid); - measureCodec = new BufferedMeasureEncoder(cubeDesc.getMeasures()); + measureCodec = new BufferedMeasureCodec(cubeDesc.getMeasures()); measures = new Object[cubeDesc.getMeasures().size()]; int colCount = cubeDesc.getRowkey().getRowKeyColumns().length; http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 6986344..d6e1d7e 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 @@ -29,7 +29,7 @@ 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.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureAggregators; import org.apache.kylin.metadata.model.MeasureDesc; import org.slf4j.Logger; @@ -47,7 +47,7 @@ public class CuboidReducer extends KylinReducer<Text, Text, Text, Text> { private CubeDesc cubeDesc; private List<MeasureDesc> measuresDescs; - private BufferedMeasureEncoder codec; + private BufferedMeasureCodec codec; private MeasureAggregators aggs; private int counter; @@ -71,7 +71,7 @@ public class CuboidReducer extends KylinReducer<Text, Text, Text, Text> { cubeDesc = CubeManager.getInstance(config).getCube(cubeName).getDescriptor(); measuresDescs = cubeDesc.getMeasures(); - codec = new BufferedMeasureEncoder(measuresDescs); + codec = new BufferedMeasureCodec(measuresDescs); aggs = new MeasureAggregators(measuresDescs); input = new Object[measuresDescs.size()]; http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 c525e90..5e278f8 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 @@ -28,7 +28,7 @@ import org.apache.kylin.common.util.ByteArray; import org.apache.kylin.common.util.Bytes; import org.apache.kylin.cube.cuboid.CuboidScheduler; import org.apache.kylin.engine.mr.common.BatchConstants; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter; import com.google.common.collect.Lists; @@ -159,7 +159,7 @@ public class FactDistinctHiveColumnsMapper<KEYIN> extends FactDistinctColumnsMap @Override protected void cleanup(Context context) throws IOException, InterruptedException { if (collectStatistics) { - ByteBuffer hllBuf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE); + ByteBuffer hllBuf = ByteBuffer.allocate(BufferedMeasureCodec.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/fa5c9cb7/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 54b7b7f..cfecf23 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 @@ -31,7 +31,7 @@ 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.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureAggregators; import org.apache.kylin.metadata.model.MeasureDesc; import org.slf4j.Logger; @@ -43,7 +43,7 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra private static final Logger logger = LoggerFactory.getLogger(InMemCuboidReducer.class); - private BufferedMeasureEncoder codec; + private BufferedMeasureCodec codec; private MeasureAggregators aggs; private int counter; @@ -63,7 +63,7 @@ public class InMemCuboidReducer extends KylinReducer<ByteArrayWritable, ByteArra CubeDesc cubeDesc = cube.getDescriptor(); List<MeasureDesc> measuresDescs = cubeDesc.getMeasures(); - codec = new BufferedMeasureEncoder(measuresDescs); + codec = new BufferedMeasureCodec(measuresDescs); aggs = new MeasureAggregators(measuresDescs); input = new Object[measuresDescs.size()]; result = new Object[measuresDescs.size()]; http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 0c4fa56..aa323fd 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 @@ -30,7 +30,7 @@ import org.apache.kylin.cube.kv.AbstractRowKeyEncoder; import org.apache.kylin.cube.model.CubeDesc; import org.apache.kylin.engine.mr.ByteArrayWritable; import org.apache.kylin.gridtable.GTRecord; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -49,7 +49,7 @@ public abstract class KVGTRecordWriter implements ICuboidWriter { private int measureCount; private byte[] keyBuf; private ImmutableBitSet measureColumns; - private ByteBuffer valueBuf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE); + private ByteBuffer valueBuf = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); private ByteArrayWritable outputKey = new ByteArrayWritable(); private ByteArrayWritable outputValue = new ByteArrayWritable(); private long cuboidRowCount = 0; http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 a6bef83..55b8474 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.dict.DictionaryManager; 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.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureIngester; import org.apache.kylin.measure.MeasureType; import org.apache.kylin.metadata.model.MeasureDesc; @@ -85,7 +85,7 @@ 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 BufferedMeasureEncoder codec; + private BufferedMeasureCodec codec; private Object[] measureObjs; private Text outputValue; @@ -115,7 +115,7 @@ public class MergeCuboidMapper extends KylinMapper<Text, Text, Text, Text> { rowKeyEncoderProvider = new RowKeyEncoderProvider(mergedCubeSegment); measureDescs = cubeDesc.getMeasures(); - codec = new BufferedMeasureEncoder(measureDescs); + codec = new BufferedMeasureCodec(measureDescs); measureObjs = new Object[measureDescs.size()]; outputValue = new Text(); http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 3ca09cf..5e687a5 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 @@ -36,7 +36,7 @@ import org.apache.kylin.common.util.LocalFileMetadataTestCase; import org.apache.kylin.cube.CubeManager; 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.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureAggregator; import org.apache.kylin.measure.MeasureIngester; import org.apache.kylin.measure.MeasureType; @@ -78,7 +78,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(); - BufferedMeasureEncoder codec = new BufferedMeasureEncoder(cubeDesc.getMeasures()); + BufferedMeasureCodec codec = new BufferedMeasureCodec(cubeDesc.getMeasures()); Text key1 = new Text("72010ustech"); List<Text> values1 = new ArrayList<Text>(); @@ -125,7 +125,7 @@ public class CubeReducerTest extends LocalFileMetadataTestCase { MeasureType origMeasureType = functionDesc.getMeasureType(); field.set(functionDesc, new MockUpMeasureType(origMeasureType)); - BufferedMeasureEncoder codec = new BufferedMeasureEncoder(cubeDesc.getMeasures()); + BufferedMeasureCodec codec = new BufferedMeasureCodec(cubeDesc.getMeasures()); Text key1 = new Text("72010ustech"); List<Text> values1 = new ArrayList<Text>(); @@ -159,7 +159,7 @@ public class CubeReducerTest extends LocalFileMetadataTestCase { assertTrue(result.contains(p3)); } - private Text newValueText(BufferedMeasureEncoder codec, String sum, String min, String max, int count, int item_count) { + private Text newValueText(BufferedMeasureCodec 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) }; ByteBuffer buf = codec.encode(values); http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 5c2def2..9b69063 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 @@ -81,7 +81,7 @@ 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.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureAggregators; import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter; import org.apache.kylin.metadata.model.FunctionDesc; @@ -424,7 +424,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 BufferedMeasureEncoder codec = new BufferedMeasureEncoder(dataTypes); + final BufferedMeasureCodec codec = new BufferedMeasureCodec(dataTypes); final Object[] input = new Object[measureSize]; final Object[] result = new Object[measureSize]; http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 ea08f73..4f407ff 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 @@ -29,7 +29,7 @@ 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.measure.BufferedMeasureCodec; import org.apache.kylin.metadata.model.TblColRef; import scala.Tuple2; @@ -54,7 +54,7 @@ public final class DefaultTupleConverter implements TupleConverter { private ByteBuffer getValueBuf() { if (valueBuf.get() == null) { - valueBuf.set(ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE)); + valueBuf.set(ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE)); } return valueBuf.get(); } http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 c90c62d..f27bee3 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 @@ -34,7 +34,7 @@ import org.apache.kylin.engine.mr.KylinMapper; import org.apache.kylin.engine.mr.MRUtil; 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.BufferedMeasureCodec; import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter; import org.apache.kylin.metadata.MetadataManager; import org.apache.kylin.metadata.model.ColumnDesc; @@ -97,7 +97,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(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE); + ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureCodec.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/fa5c9cb7/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 1a7a253..858d84c 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 @@ -31,7 +31,7 @@ import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.kylin.engine.mr.KylinReducer; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter; /** @@ -79,7 +79,7 @@ public class ColumnCardinalityReducer extends KylinReducer<IntWritable, BytesWri while (it.hasNext()) { int key = it.next(); HyperLogLogPlusCounter hllc = hllcMap.get(key); - ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE); + ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); buf.clear(); hllc.writeRegisters(buf); buf.flip(); http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 1e45f64..d27860a 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 @@ -34,7 +34,7 @@ import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.mrunit.mapreduce.ReduceDriver; import org.apache.hadoop.mrunit.types.Pair; import org.apache.kylin.common.util.Bytes; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.hllc.HyperLogLogPlusCounter; import org.junit.Before; import org.junit.Test; @@ -65,7 +65,7 @@ public class ColumnCardinalityReducerTest { i++; hllc.add(Bytes.toBytes(temp)); } - ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE); + ByteBuffer buf = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); buf.clear(); hllc.writeRegisters(buf); buf.flip(); http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 9ed15c6..29a30c1 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 @@ -29,7 +29,7 @@ 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.model.HBaseColumnDesc; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.measure.MeasureAggregator; import org.apache.kylin.measure.MeasureType; import org.apache.kylin.measure.MeasureTypeFactory; @@ -241,7 +241,7 @@ public class ObserverAggregators { final String[] dataTypes; final int nMeasures; - final BufferedMeasureEncoder measureCodec; + final BufferedMeasureCodec measureCodec; final Object[] measureValues; public HCol(byte[] bFamily, byte[] bQualifier, String[] funcNames, String[] dataTypes) { @@ -252,7 +252,7 @@ public class ObserverAggregators { this.nMeasures = funcNames.length; assert funcNames.length == dataTypes.length; - this.measureCodec = new BufferedMeasureEncoder(dataTypes); + this.measureCodec = new BufferedMeasureCodec(dataTypes); this.measureValues = new Object[nMeasures]; } http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 13a7b53..4790d6e 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 @@ -55,7 +55,7 @@ import org.apache.kylin.gridtable.GTScanTimeoutException; import org.apache.kylin.gridtable.IGTScanner; import org.apache.kylin.gridtable.IGTStore; import org.apache.kylin.gridtable.StorageSideBehavior; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.metadata.filter.UDF.MassInTupleFilter; import org.apache.kylin.metadata.model.TblColRef; import org.apache.kylin.metadata.realization.IRealizationConstants; @@ -290,9 +290,9 @@ public class CubeVisitService extends CubeVisitProtos.CubeVisitService implement IGTScanner rawScanner = store.scan(scanReq); IGTScanner finalScanner = scanReq.decorateScanner(rawScanner, behavior.filterToggledOn(), behavior.aggrToggledOn(), deadline); - ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE); + ByteBuffer buffer = ByteBuffer.allocate(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE); - ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureEncoder.DEFAULT_BUFFER_SIZE);//ByteArrayOutputStream will auto grow + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(BufferedMeasureCodec.DEFAULT_BUFFER_SIZE);//ByteArrayOutputStream will auto grow int finalRowCount = 0; try { http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 7523249..8205ff7 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.MeasureDecoder; +import org.apache.kylin.measure.MeasureCodec; import com.google.common.collect.Lists; @@ -48,7 +48,7 @@ public class CubeHFileMapper extends KylinMapper<Text, Text, ImmutableBytesWrita String cubeName; CubeDesc cubeDesc; - MeasureDecoder inputCodec; + MeasureCodec 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 MeasureDecoder(cubeDesc.getMeasures()); + inputCodec = new MeasureCodec(cubeDesc.getMeasures()); inputMeasures = new Object[cubeDesc.getMeasures().size()]; keyValueCreators = Lists.newArrayList(); http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 490031e..222c9f4 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 @@ -26,7 +26,7 @@ import org.apache.hadoop.io.Text; import org.apache.kylin.common.util.Bytes; import org.apache.kylin.cube.model.CubeDesc; import org.apache.kylin.cube.model.HBaseColumnDesc; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.metadata.model.MeasureDesc; /** @@ -40,7 +40,7 @@ public class KeyValueCreator { int[] refIndex; MeasureDesc[] refMeasures; - BufferedMeasureEncoder codec; + BufferedMeasureCodec codec; Object[] colValues; public boolean isFullCopy; @@ -54,7 +54,7 @@ public class KeyValueCreator { refIndex = colDesc.getMeasureIndex(); refMeasures = colDesc.getMeasures(); - codec = new BufferedMeasureEncoder(refMeasures); + codec = new BufferedMeasureCodec(refMeasures); colValues = new Object[refMeasures.length]; isFullCopy = true; http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 2aa285c..86104e2 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.MeasureDecoder; +import org.apache.kylin.measure.MeasureCodec; 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 MeasureDecoder codec; + private final MeasureCodec 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 MeasureDecoder(measures); + this.codec = new MeasureCodec(measures); this.values = new Object[measures.length]; } http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 c1fd2e2..77b18e2 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 @@ -33,7 +33,7 @@ import org.apache.kylin.common.util.LocalFileMetadataTestCase; import org.apache.kylin.cube.CubeManager; import org.apache.kylin.cube.model.CubeDesc; import org.apache.kylin.engine.mr.HadoopUtil; -import org.apache.kylin.measure.MeasureDecoder; +import org.apache.kylin.measure.MeasureCodec; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -47,7 +47,7 @@ public class CubeHFileMapper2Test extends LocalFileMetadataTestCase { String cubeName = "test_kylin_cube_with_slr_ready"; - MeasureDecoder codec; + MeasureCodec codec; Object[] outKV = new Object[2]; @Before @@ -57,7 +57,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 MeasureDecoder(desc.getMeasures()); + codec = new MeasureCodec(desc.getMeasures()); } @After http://git-wip-us.apache.org/repos/asf/kylin/blob/fa5c9cb7/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 81ac32f..f0ad6e0 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 @@ -29,7 +29,7 @@ import org.apache.kylin.common.util.LocalFileMetadataTestCase; import org.apache.kylin.cube.CubeManager; import org.apache.kylin.cube.model.CubeDesc; import org.apache.kylin.cube.model.HBaseColumnDesc; -import org.apache.kylin.measure.BufferedMeasureEncoder; +import org.apache.kylin.measure.BufferedMeasureCodec; import org.apache.kylin.metadata.MetadataManager; import org.apache.kylin.metadata.datatype.LongMutable; import org.apache.kylin.metadata.model.FunctionDesc; @@ -56,7 +56,7 @@ 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]; - BufferedMeasureEncoder codec = new BufferedMeasureEncoder(hbaseCol.getMeasures()); + BufferedMeasureCodec codec = new BufferedMeasureCodec(hbaseCol.getMeasures()); BigDecimal sum = new BigDecimal("333.1234567"); BigDecimal min = new BigDecimal("333.1111111"); BigDecimal max = new BigDecimal("333.1999999"); @@ -86,7 +86,7 @@ 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]; - BufferedMeasureEncoder codec = new BufferedMeasureEncoder(hbaseCol.getMeasures()); + BufferedMeasureCodec codec = new BufferedMeasureCodec(hbaseCol.getMeasures()); BigDecimal sum = new BigDecimal("11111111111111111111333.1234567"); BigDecimal min = new BigDecimal("333.1111111"); BigDecimal max = new BigDecimal("333.1999999");
