| /* |
| * 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.carbondata.core.datastore.page.encoding.compress; |
| |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.BitSet; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.carbondata.core.constants.CarbonCommonConstants; |
| import org.apache.carbondata.core.datastore.ReusableDataBuffer; |
| import org.apache.carbondata.core.datastore.TableSpec; |
| import org.apache.carbondata.core.datastore.compression.Compressor; |
| import org.apache.carbondata.core.datastore.compression.CompressorFactory; |
| import org.apache.carbondata.core.datastore.page.ColumnPage; |
| import org.apache.carbondata.core.datastore.page.ColumnPageValueConverter; |
| import org.apache.carbondata.core.datastore.page.LazyColumnPage; |
| import org.apache.carbondata.core.datastore.page.VarLengthColumnPageBase; |
| import org.apache.carbondata.core.datastore.page.encoding.ColumnPageCodec; |
| import org.apache.carbondata.core.datastore.page.encoding.ColumnPageDecoder; |
| import org.apache.carbondata.core.datastore.page.encoding.ColumnPageEncoder; |
| import org.apache.carbondata.core.datastore.page.encoding.ColumnPageEncoderMeta; |
| import org.apache.carbondata.core.keygenerator.directdictionary.timestamp.DateDirectDictionaryGenerator; |
| import org.apache.carbondata.core.metadata.datatype.DataType; |
| import org.apache.carbondata.core.metadata.datatype.DataTypes; |
| import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory; |
| import org.apache.carbondata.core.metadata.datatype.DecimalType; |
| import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector; |
| import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo; |
| import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl; |
| import org.apache.carbondata.core.scan.result.vector.impl.directread.ColumnarVectorWrapperDirectFactory; |
| import org.apache.carbondata.core.scan.result.vector.impl.directread.ConvertibleVector; |
| import org.apache.carbondata.core.scan.result.vector.impl.directread.SequentialFill; |
| import org.apache.carbondata.core.util.ByteUtil; |
| import org.apache.carbondata.format.Encoding; |
| |
| /** |
| * This codec directly apply compression on the input data |
| */ |
| public class DirectCompressCodec implements ColumnPageCodec { |
| |
| private DataType dataType; |
| |
| public DirectCompressCodec(DataType dataType) { |
| this.dataType = dataType; |
| } |
| |
| boolean isComplexPrimitiveIntLengthEncoding = false; |
| |
| public void setComplexPrimitiveIntLengthEncoding(boolean complexPrimitiveIntLengthEncoding) { |
| isComplexPrimitiveIntLengthEncoding = complexPrimitiveIntLengthEncoding; |
| } |
| |
| @Override |
| public String getName() { |
| return "DirectCompressCodec"; |
| } |
| |
| @Override |
| public ColumnPageEncoder createEncoder(Map<String, String> parameter) { |
| return new ColumnPageEncoder() { |
| |
| @Override |
| protected ByteBuffer encodeData(ColumnPage input) throws IOException { |
| Compressor compressor = CompressorFactory.getInstance().getCompressor( |
| input.getColumnCompressorName()); |
| return input.compress(compressor); |
| } |
| |
| @Override |
| protected List<Encoding> getEncodingList() { |
| List<Encoding> encodings = new ArrayList<>(); |
| encodings.add(dataType == DataTypes.VARCHAR ? |
| Encoding.DIRECT_COMPRESS_VARCHAR : |
| Encoding.DIRECT_COMPRESS); |
| return encodings; |
| } |
| |
| @Override |
| protected ColumnPageEncoderMeta getEncoderMeta(ColumnPage inputPage) { |
| return new ColumnPageEncoderMeta(inputPage.getColumnSpec(), inputPage.getDataType(), |
| inputPage.getStatistics(), inputPage.getColumnCompressorName()); |
| } |
| }; |
| } |
| |
| @Override |
| public ColumnPageDecoder createDecoder(final ColumnPageEncoderMeta meta) { |
| return new ColumnPageDecoder() { |
| |
| @Override |
| public ColumnPage decode(byte[] input, int offset, int length) { |
| ColumnPage decodedPage; |
| if (DataTypes.isDecimal(dataType)) { |
| decodedPage = ColumnPage.decompressDecimalPage(meta, input, offset, length); |
| } else { |
| decodedPage = ColumnPage |
| .decompress(meta, input, offset, length, false, isComplexPrimitiveIntLengthEncoding); |
| } |
| return LazyColumnPage.newPage(decodedPage, converter); |
| } |
| |
| @Override |
| public void decodeAndFillVector(byte[] input, int offset, int length, |
| ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize, |
| ReusableDataBuffer reusableDataBuffer) { |
| Compressor compressor = |
| CompressorFactory.getInstance().getCompressor(meta.getCompressorName()); |
| int uncompressedLength; |
| byte[] unCompressData; |
| if (null != reusableDataBuffer && compressor.supportReusableBuffer()) { |
| uncompressedLength = compressor.unCompressedLength(input, offset, length); |
| unCompressData = reusableDataBuffer.getDataBuffer(uncompressedLength); |
| compressor.rawUncompress(input, offset, length, unCompressData); |
| } else { |
| unCompressData = compressor.unCompressByte(input, offset, length); |
| uncompressedLength = unCompressData.length; |
| } |
| if (DataTypes.isDecimal(dataType)) { |
| TableSpec.ColumnSpec columnSpec = meta.getColumnSpec(); |
| DecimalConverterFactory.DecimalConverter decimalConverter = |
| DecimalConverterFactory.INSTANCE |
| .getDecimalConverter(columnSpec.getPrecision(), columnSpec.getScale()); |
| vectorInfo.decimalConverter = decimalConverter; |
| if (DataTypes.isDecimal(meta.getStoreDataType())) { |
| ColumnPage decimalColumnPage = VarLengthColumnPageBase |
| .newDecimalColumnPage(meta, unCompressData, uncompressedLength); |
| decimalConverter.fillVector(decimalColumnPage.getByteArrayPage(), pageSize, vectorInfo, |
| nullBits, meta.getStoreDataType()); |
| } else { |
| converter |
| .decodeAndFillVector(unCompressData, vectorInfo, nullBits, meta.getStoreDataType(), |
| pageSize); |
| } |
| } else { |
| converter |
| .decodeAndFillVector(unCompressData, vectorInfo, nullBits, meta.getStoreDataType(), |
| pageSize); |
| } |
| } |
| |
| @Override |
| public ColumnPage decode(byte[] input, int offset, int length, boolean isLVEncoded) { |
| return LazyColumnPage.newPage(ColumnPage |
| .decompress(meta, input, offset, length, isLVEncoded, |
| isComplexPrimitiveIntLengthEncoding), converter); |
| } |
| }; |
| } |
| |
| private ColumnPageValueConverter converter = new ColumnPageValueConverter() { |
| @Override |
| public void encode(int rowId, byte value) { |
| throw new RuntimeException("internal error"); |
| } |
| |
| @Override |
| public void encode(int rowId, short value) { |
| throw new RuntimeException("internal error"); |
| } |
| |
| @Override |
| public void encode(int rowId, int value) { |
| throw new RuntimeException("internal error"); |
| } |
| |
| @Override |
| public void encode(int rowId, long value) { |
| throw new RuntimeException("internal error"); |
| } |
| |
| @Override |
| public void encode(int rowId, float value) { |
| throw new RuntimeException("internal error"); |
| } |
| |
| @Override |
| public void encode(int rowId, double value) { |
| throw new RuntimeException("internal error"); |
| } |
| |
| @Override |
| public long decodeLong(byte value) { |
| return value; |
| } |
| |
| @Override |
| public long decodeLong(short value) { |
| return value; |
| } |
| |
| @Override |
| public long decodeLong(int value) { |
| return value; |
| } |
| |
| @Override |
| public double decodeDouble(byte value) { |
| return value; |
| } |
| |
| @Override |
| public double decodeDouble(short value) { |
| return value; |
| } |
| |
| @Override |
| public double decodeDouble(int value) { |
| return value; |
| } |
| |
| @Override |
| public double decodeDouble(long value) { |
| return value; |
| } |
| |
| @Override |
| public double decodeDouble(float value) { |
| return value; |
| } |
| |
| @Override |
| public double decodeDouble(double value) { |
| return value; |
| } |
| |
| @Override |
| public void decodeAndFillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits, |
| DataType pageDataType, int pageSize) { |
| CarbonColumnVector vector = vectorInfo.vector; |
| DataType vectorDataType = vector.getType(); |
| BitSet deletedRows = vectorInfo.deletedRows; |
| if (vectorDataType.isComplexType() && vectorInfo.vectorStack.isEmpty()) { |
| // Only if vectorStack is empty, initialize with the parent vector |
| vectorInfo.vectorStack.push(vectorInfo.vector); |
| } |
| // If top of vector stack is a complex vector, |
| // then add their children into the stack and load them too. |
| if (!vectorInfo.vectorStack.isEmpty() && vectorInfo.vectorStack.peek().getType() |
| .isComplexType()) { |
| CarbonColumnVector parentVector = vectorInfo.vectorStack.peek(); |
| CarbonColumnVectorImpl parentVectorImpl = |
| (CarbonColumnVectorImpl) (parentVector.getColumnVector()); |
| // parse the parent page data, |
| // save the information about number of child in each row in parent vector |
| if (DataTypes.isStructType(parentVectorImpl.getType())) { |
| parentVectorImpl.setNumberOfChildElementsForStruct(pageData, pageSize); |
| } else { |
| parentVectorImpl.setNumberOfChildElementsForArray(pageData, pageSize); |
| } |
| for (CarbonColumnVector childVector : parentVector.getColumnVector().getChildrenVector()) { |
| // push each child |
| vectorInfo.vectorStack.push(childVector); |
| // load the child page, here child page loading flow updated vector from top of the stack |
| // and pop() the child vector once loading is finished. |
| ((CarbonColumnVectorImpl) (vectorInfo.vectorStack.peek().getColumnVector())).loadPage(); |
| } |
| vector = ColumnarVectorWrapperDirectFactory |
| .getDirectVectorWrapperFactory(vectorInfo, parentVector, vectorInfo.invertedIndex, |
| nullBits, vectorInfo.deletedRows, true, false); |
| fillVectorBasedOnType(pageData, vector, vectorDataType, pageDataType, pageSize, |
| vectorInfo, nullBits); |
| } else { |
| pageSize = ColumnVectorInfo.getUpdatedPageSizeForChildVector(vectorInfo, pageSize); |
| vector = ColumnarVectorWrapperDirectFactory |
| .getDirectVectorWrapperFactory(vectorInfo, vector, vectorInfo.invertedIndex, nullBits, |
| deletedRows, true, false); |
| fillVectorBasedOnType(pageData, vector, vector.getType(), pageDataType, pageSize, |
| vectorInfo, nullBits); |
| if ((deletedRows == null || deletedRows.isEmpty()) |
| && !(vectorInfo.vector instanceof SequentialFill)) { |
| for (int i = nullBits.nextSetBit(0); i >= 0; i = nullBits.nextSetBit(i + 1)) { |
| vector.putNull(i); |
| } |
| } |
| if (vector instanceof ConvertibleVector) { |
| ((ConvertibleVector) vector).convert(); |
| } |
| } |
| } |
| |
| private void fillVectorBasedOnType(byte[] pageData, CarbonColumnVector vector, |
| DataType vectorDataType, DataType pageDataType, int pageSize, ColumnVectorInfo vectorInfo, |
| BitSet nullBits) { |
| if (vectorInfo.vector.getColumnVector() != null |
| && vector.getColumnVector() == vectorInfo.vector.getColumnVector() && vectorInfo.vector |
| .getColumnVector().getType().isComplexType()) { |
| List<Integer> childElementsForEachRow = |
| ((CarbonColumnVectorImpl) vector.getColumnVector()) |
| .getNumberOfChildrenElementsInEachRow(); |
| vector.getColumnVector().putComplexObject(childElementsForEachRow); |
| } else { |
| fillPrimitiveType(pageData, vector, vectorDataType, pageDataType, pageSize, vectorInfo, |
| nullBits); |
| } |
| } |
| |
| private void fillPrimitiveType(byte[] pageData, CarbonColumnVector vector, |
| DataType vectorDataType, DataType pageDataType, int pageSize, ColumnVectorInfo vectorInfo, |
| BitSet nullBits) { |
| int rowId = 0; |
| if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) { |
| if (vectorDataType == DataTypes.SHORT) { |
| for (int i = 0; i < pageSize; i++) { |
| vector.putShort(i, pageData[i]); |
| } |
| } else if (vectorDataType == DataTypes.INT) { |
| for (int i = 0; i < pageSize; i++) { |
| vector.putInt(i, pageData[i]); |
| } |
| } else if (vectorDataType == DataTypes.LONG) { |
| for (int i = 0; i < pageSize; i++) { |
| vector.putLong(i, pageData[i]); |
| } |
| } else if (vectorDataType == DataTypes.TIMESTAMP) { |
| for (int i = 0; i < pageSize; i++) { |
| vector.putLong(i, (long) pageData[i] * 1000); |
| } |
| } else if (vectorDataType == DataTypes.BOOLEAN || vectorDataType == DataTypes.BYTE) { |
| vector.putBytes(0, pageSize, pageData, 0); |
| } else if (DataTypes.isDecimal(vectorDataType)) { |
| DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; |
| decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); |
| } else { |
| for (int i = 0; i < pageSize; i++) { |
| vector.putDouble(i, pageData[i]); |
| } |
| } |
| } else if (pageDataType == DataTypes.SHORT) { |
| int shortSizeInBytes = DataTypes.SHORT.getSizeInBytes(); |
| int size = pageSize * shortSizeInBytes; |
| if (vectorDataType == DataTypes.SHORT) { |
| for (int i = 0; i < size; i += shortSizeInBytes) { |
| vector.putShort(rowId++, (ByteUtil.toShortLittleEndian(pageData, i))); |
| } |
| } else if (vectorDataType == DataTypes.INT) { |
| for (int i = 0; i < size; i += shortSizeInBytes) { |
| vector.putInt(rowId++, ByteUtil.toShortLittleEndian(pageData, i)); |
| } |
| } else if (vectorDataType == DataTypes.LONG) { |
| for (int i = 0; i < size; i += shortSizeInBytes) { |
| vector.putLong(rowId++, ByteUtil.toShortLittleEndian(pageData, i)); |
| } |
| } else if (vectorDataType == DataTypes.TIMESTAMP) { |
| for (int i = 0; i < size; i += shortSizeInBytes) { |
| vector.putLong(rowId++, (long) ByteUtil.toShortLittleEndian(pageData, i) * 1000); |
| } |
| } else if (DataTypes.isDecimal(vectorDataType)) { |
| DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; |
| decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); |
| } else { |
| for (int i = 0; i < size; i += shortSizeInBytes) { |
| vector.putDouble(rowId++, ByteUtil.toShortLittleEndian(pageData, i)); |
| } |
| } |
| } else if (pageDataType == DataTypes.SHORT_INT) { |
| if (vectorDataType == DataTypes.INT) { |
| for (int i = 0; i < pageSize; i++) { |
| int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); |
| vector.putInt(i, shortInt); |
| } |
| } else if (vectorDataType == DataTypes.LONG) { |
| for (int i = 0; i < pageSize; i++) { |
| int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); |
| vector.putLong(i, shortInt); |
| } |
| } else if (vectorDataType == DataTypes.TIMESTAMP) { |
| for (int i = 0; i < pageSize; i++) { |
| int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); |
| vector.putLong(i, (long) shortInt * 1000); |
| } |
| } else if (DataTypes.isDecimal(vectorDataType)) { |
| DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; |
| decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); |
| } else { |
| for (int i = 0; i < pageSize; i++) { |
| int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); |
| vector.putDouble(i, shortInt); |
| } |
| } |
| } else { |
| int intSizeInBytes = DataTypes.INT.getSizeInBytes(); |
| if (pageDataType == DataTypes.INT) { |
| int size = pageSize * intSizeInBytes; |
| if (vectorDataType == DataTypes.INT) { |
| for (int i = 0; i < size; i += intSizeInBytes) { |
| vector.putInt(rowId++, ByteUtil.toIntLittleEndian(pageData, i)); |
| } |
| } else if (vectorDataType == DataTypes.LONG) { |
| for (int i = 0; i < size; i += intSizeInBytes) { |
| vector.putLong(rowId++, ByteUtil.toIntLittleEndian(pageData, i)); |
| } |
| } else if (vectorDataType == DataTypes.TIMESTAMP) { |
| for (int i = 0; i < size; i += intSizeInBytes) { |
| vector.putLong(rowId++, (long) ByteUtil.toIntLittleEndian(pageData, i) * 1000); |
| } |
| } else if (DataTypes.isDecimal(vectorDataType)) { |
| DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; |
| decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); |
| } else { |
| for (int i = 0; i < size; i += intSizeInBytes) { |
| vector.putDouble(rowId++, ByteUtil.toIntLittleEndian(pageData, i)); |
| } |
| } |
| } else if (pageDataType == DataTypes.LONG) { |
| int longSizeInBytes = DataTypes.LONG.getSizeInBytes(); |
| int size = pageSize * longSizeInBytes; |
| if (vectorDataType == DataTypes.LONG) { |
| for (int i = 0; i < size; i += longSizeInBytes) { |
| vector.putLong(rowId++, ByteUtil.toLongLittleEndian(pageData, i)); |
| } |
| } else if (vectorDataType == DataTypes.TIMESTAMP) { |
| for (int i = 0; i < size; i += longSizeInBytes) { |
| vector.putLong(rowId++, ByteUtil.toLongLittleEndian(pageData, i) * 1000); |
| } |
| } else if (DataTypes.isDecimal(vectorDataType)) { |
| DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; |
| decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); |
| } |
| } else if (pageDataType == DataTypes.BYTE_ARRAY) { |
| // for complex primitive types |
| if (vectorDataType == DataTypes.STRING || vectorDataType == DataTypes.BINARY |
| || vectorDataType == DataTypes.VARCHAR) { |
| // for complex primitive string, binary, varchar type |
| int offset = 0; |
| for (int i = 0; i < pageSize; i++) { |
| int len = ByteBuffer.wrap(pageData, offset, intSizeInBytes).getInt(); |
| offset += intSizeInBytes; |
| if (vectorDataType == DataTypes.BINARY && len == 0) { |
| vector.putNull(i); |
| continue; |
| } |
| byte[] row = new byte[len]; |
| System.arraycopy(pageData, offset, row, 0, len); |
| if (Arrays.equals(row, CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY)) { |
| vector.putNull(i); |
| } else { |
| vector.putObject(i, row); |
| } |
| offset += len; |
| } |
| } else if (vectorDataType == DataTypes.DATE) { |
| // for complex primitive date type |
| int offset = 0; |
| for (int i = 0; i < pageSize; i++) { |
| int len = ByteBuffer.wrap(pageData, offset, intSizeInBytes).getInt(); |
| offset += intSizeInBytes; |
| int surrogateInternal = |
| ByteUtil.toXorInt(pageData, offset, intSizeInBytes); |
| if (len == 0) { |
| vector.putObject(0, null); |
| } else { |
| vector.putObject(0, surrogateInternal - DateDirectDictionaryGenerator.cutOffDate); |
| } |
| offset += len; |
| } |
| } else if (DataTypes.isDecimal(vectorDataType)) { |
| // for complex primitive decimal type |
| DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; |
| if (decimalConverter == null) { |
| decimalConverter = DecimalConverterFactory.INSTANCE |
| .getDecimalConverter(((DecimalType) vectorDataType).getPrecision(), |
| ((DecimalType) vectorDataType).getScale()); |
| } |
| decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); |
| } |
| } else if (vectorDataType == DataTypes.FLOAT) { |
| int floatSizeInBytes = DataTypes.FLOAT.getSizeInBytes(); |
| int size = pageSize * floatSizeInBytes; |
| for (int i = 0; i < size; i += floatSizeInBytes) { |
| vector.putFloat(rowId++, ByteUtil.toFloatLittleEndian(pageData, i)); |
| } |
| } else { |
| int doubleSizeInBytes = DataTypes.DOUBLE.getSizeInBytes(); |
| int size = pageSize * doubleSizeInBytes; |
| for (int i = 0; i < size; i += doubleSizeInBytes) { |
| vector.putDouble(rowId++, ByteUtil.toDoubleLittleEndian(pageData, i)); |
| } |
| } |
| } |
| } |
| }; |
| |
| } |
