| /* |
| * 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.hadoop.stream; |
| |
| import java.io.IOException; |
| import java.math.BigDecimal; |
| import java.nio.ByteBuffer; |
| import java.util.BitSet; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.carbondata.core.constants.CarbonCommonConstants; |
| import org.apache.carbondata.core.datastore.block.SegmentProperties; |
| import org.apache.carbondata.core.datastore.compression.CompressorFactory; |
| import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryGenerator; |
| import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory; |
| import org.apache.carbondata.core.metadata.blocklet.index.BlockletMinMaxIndex; |
| import org.apache.carbondata.core.metadata.datatype.DataType; |
| import org.apache.carbondata.core.metadata.datatype.DataTypes; |
| import org.apache.carbondata.core.metadata.schema.table.CarbonTable; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure; |
| import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema; |
| import org.apache.carbondata.core.reader.CarbonHeaderReader; |
| import org.apache.carbondata.core.scan.expression.exception.FilterUnsupportedException; |
| import org.apache.carbondata.core.scan.filter.FilterUtil; |
| import org.apache.carbondata.core.scan.filter.GenericQueryType; |
| import org.apache.carbondata.core.scan.filter.executer.FilterExecutor; |
| import org.apache.carbondata.core.scan.filter.intf.RowImpl; |
| import org.apache.carbondata.core.scan.filter.intf.RowIntf; |
| import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf; |
| import org.apache.carbondata.core.scan.model.QueryModel; |
| import org.apache.carbondata.core.util.CarbonMetadataUtil; |
| import org.apache.carbondata.core.util.CarbonUtil; |
| import org.apache.carbondata.core.util.DataTypeUtil; |
| import org.apache.carbondata.format.BlockletHeader; |
| import org.apache.carbondata.format.FileHeader; |
| import org.apache.carbondata.hadoop.CarbonInputSplit; |
| import org.apache.carbondata.hadoop.CarbonMultiBlockSplit; |
| import org.apache.carbondata.hadoop.api.CarbonTableInputFormat; |
| import org.apache.carbondata.processing.util.CarbonDataProcessorUtil; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FSDataInputStream; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.mapreduce.InputSplit; |
| import org.apache.hadoop.mapreduce.RecordReader; |
| import org.apache.hadoop.mapreduce.TaskAttemptContext; |
| import org.apache.hadoop.mapreduce.lib.input.FileSplit; |
| |
| /** |
| * Stream row record reader |
| */ |
| public class StreamRecordReader extends RecordReader<Void, Object> { |
| |
| // metadata |
| protected CarbonTable carbonTable; |
| private CarbonColumn[] storageColumns; |
| private boolean[] isRequired; |
| private boolean[] dimensionsIsVarcharTypeMap; |
| private DataType[] measureDataTypes; |
| private int dimensionCount; |
| private int measureCount; |
| |
| // input |
| private FileSplit fileSplit; |
| private Configuration hadoopConf; |
| protected StreamBlockletReader input; |
| protected boolean isFirstRow = true; |
| protected QueryModel model; |
| |
| // decode data |
| private BitSet allNonNull; |
| private boolean[] isNoDictColumn; |
| private DirectDictionaryGenerator[] directDictionaryGenerators; |
| private GenericQueryType[] queryTypes; |
| private String compressorName; |
| |
| // vectorized reader |
| protected boolean isFinished = false; |
| |
| // filter |
| protected FilterExecutor filter; |
| private boolean[] isFilterRequired; |
| private Object[] filterValues; |
| protected RowIntf filterRow; |
| private int[] filterMap; |
| |
| // output |
| protected CarbonColumn[] projection; |
| private boolean[] isProjectionRequired; |
| private int[] projectionMap; |
| protected Object[] outputValues; |
| |
| // empty project, null filter |
| protected boolean skipScanData; |
| |
| // return raw row for hand off |
| private final boolean useRawRow; |
| |
| public StreamRecordReader(QueryModel mdl, boolean useRawRow) { |
| this.model = mdl; |
| this.useRawRow = useRawRow; |
| |
| } |
| |
| @Override |
| public void initialize(InputSplit split, TaskAttemptContext context) |
| throws IOException { |
| // input |
| if (split instanceof CarbonInputSplit) { |
| fileSplit = (CarbonInputSplit) split; |
| } else if (split instanceof CarbonMultiBlockSplit) { |
| fileSplit = ((CarbonMultiBlockSplit) split).getAllSplits().get(0); |
| } else { |
| fileSplit = (FileSplit) split; |
| } |
| |
| // metadata |
| hadoopConf = context.getConfiguration(); |
| if (model == null) { |
| CarbonTableInputFormat<Object> format = new CarbonTableInputFormat<>(); |
| model = format.createQueryModel(split, context); |
| } |
| carbonTable = model.getTable(); |
| List<CarbonDimension> dimensions = carbonTable.getVisibleDimensions(); |
| dimensionCount = dimensions.size(); |
| List<CarbonMeasure> measures = carbonTable.getVisibleMeasures(); |
| measureCount = measures.size(); |
| List<CarbonColumn> carbonColumnList = carbonTable.getStreamStorageOrderColumn(); |
| storageColumns = carbonColumnList.toArray(new CarbonColumn[carbonColumnList.size()]); |
| isNoDictColumn = CarbonDataProcessorUtil.getNoDictionaryMapping(storageColumns); |
| directDictionaryGenerators = new DirectDictionaryGenerator[storageColumns.length]; |
| for (int i = 0; i < storageColumns.length; i++) { |
| if (storageColumns[i].getDataType() == DataTypes.DATE) { |
| directDictionaryGenerators[i] = DirectDictionaryKeyGeneratorFactory |
| .getDirectDictionaryGenerator(storageColumns[i].getDataType()); |
| } |
| } |
| dimensionsIsVarcharTypeMap = new boolean[dimensionCount]; |
| for (int i = 0; i < dimensionCount; i++) { |
| dimensionsIsVarcharTypeMap[i] = storageColumns[i].getDataType() == DataTypes.VARCHAR; |
| } |
| measureDataTypes = new DataType[measureCount]; |
| for (int i = 0; i < measureCount; i++) { |
| measureDataTypes[i] = storageColumns[dimensionCount + i].getDataType(); |
| } |
| |
| // decode data |
| allNonNull = new BitSet(storageColumns.length); |
| projection = model.getProjectionColumns(); |
| |
| isRequired = new boolean[storageColumns.length]; |
| boolean[] isFilterDimensions = model.getIsFilterDimensions(); |
| boolean[] isFilterMeasures = model.getIsFilterMeasures(); |
| isFilterRequired = new boolean[storageColumns.length]; |
| filterMap = new int[storageColumns.length]; |
| for (int i = 0; i < storageColumns.length; i++) { |
| if (storageColumns[i].isDimension()) { |
| if (isFilterDimensions[storageColumns[i].getOrdinal()]) { |
| isRequired[i] = true; |
| isFilterRequired[i] = true; |
| filterMap[i] = storageColumns[i].getOrdinal(); |
| } |
| } else { |
| if (isFilterMeasures[storageColumns[i].getOrdinal()]) { |
| isRequired[i] = true; |
| isFilterRequired[i] = true; |
| filterMap[i] = carbonTable.getDimensionOrdinalMax() + storageColumns[i].getOrdinal(); |
| } |
| } |
| } |
| |
| isProjectionRequired = new boolean[storageColumns.length]; |
| projectionMap = new int[storageColumns.length]; |
| for (int j = 0; j < projection.length; j++) { |
| for (int i = 0; i < storageColumns.length; i++) { |
| if (storageColumns[i].getColName().equals(projection[j].getColName())) { |
| isRequired[i] = true; |
| isProjectionRequired[i] = true; |
| projectionMap[i] = j; |
| break; |
| } |
| } |
| } |
| |
| // initialize filter |
| if (null != model.getIndexFilter()) { |
| initializeFilter(); |
| } else if (projection.length == 0) { |
| skipScanData = true; |
| } |
| |
| } |
| |
| private void initializeFilter() { |
| List<ColumnSchema> wrapperColumnSchemaList = CarbonUtil |
| .getColumnSchemaList(carbonTable.getVisibleDimensions(), carbonTable.getVisibleMeasures()); |
| |
| SegmentProperties segmentProperties = new SegmentProperties(wrapperColumnSchemaList); |
| Map<Integer, GenericQueryType> complexDimensionInfoMap = new HashMap<>(); |
| |
| FilterResolverIntf resolverIntf = model.getIndexFilter().getResolver(); |
| filter = FilterUtil.getFilterExecutorTree( |
| resolverIntf, segmentProperties, complexDimensionInfoMap, true); |
| // for row filter, we need update column index |
| FilterUtil.updateIndexOfColumnExpression(resolverIntf.getFilterExpression(), |
| carbonTable.getDimensionOrdinalMax()); |
| |
| } |
| |
| private byte[] getSyncMarker(String filePath) throws IOException { |
| CarbonHeaderReader headerReader = new CarbonHeaderReader(filePath); |
| FileHeader header = headerReader.readHeader(); |
| // legacy store does not have this member |
| if (header.isSetCompressor_name()) { |
| compressorName = header.getCompressor_name(); |
| } else { |
| compressorName = CompressorFactory.NativeSupportedCompressor.SNAPPY.getName(); |
| } |
| return header.getSync_marker(); |
| } |
| |
| protected void initializeAtFirstRow() throws IOException { |
| filterValues = new Object[carbonTable.getDimensionOrdinalMax() + measureCount]; |
| filterRow = new RowImpl(); |
| filterRow.setValues(filterValues); |
| |
| outputValues = new Object[projection.length]; |
| |
| Path file = fileSplit.getPath(); |
| |
| byte[] syncMarker = getSyncMarker(file.toString()); |
| |
| FileSystem fs = file.getFileSystem(hadoopConf); |
| |
| int bufferSize = Integer.parseInt(hadoopConf.get(CarbonStreamInputFormat.READ_BUFFER_SIZE, |
| CarbonStreamInputFormat.READ_BUFFER_SIZE_DEFAULT)); |
| |
| FSDataInputStream fileIn = fs.open(file, bufferSize); |
| fileIn.seek(fileSplit.getStart()); |
| input = new StreamBlockletReader(syncMarker, fileIn, fileSplit.getLength(), |
| fileSplit.getStart() == 0, compressorName); |
| |
| queryTypes = CarbonStreamInputFormat.getComplexDimensions(storageColumns); |
| } |
| |
| /** |
| * check next Row |
| */ |
| protected boolean nextRow() throws IOException { |
| // read row one by one |
| try { |
| boolean hasNext; |
| boolean scanMore = false; |
| do { |
| hasNext = input.hasNext(); |
| if (hasNext) { |
| if (skipScanData) { |
| input.nextRow(); |
| scanMore = false; |
| } else { |
| if (useRawRow) { |
| // read raw row for streaming hand off which does not require decode raw row |
| readRawRowFromStream(); |
| } else { |
| readRowFromStream(); |
| } |
| if (null != filter) { |
| scanMore = !filter.applyFilter(filterRow, carbonTable.getDimensionOrdinalMax()); |
| } else { |
| scanMore = false; |
| } |
| } |
| } else { |
| if (input.nextBlocklet()) { |
| BlockletHeader header = input.readBlockletHeader(); |
| if (isScanRequired(header)) { |
| if (skipScanData) { |
| input.skipBlockletData(false); |
| } else { |
| input.readBlockletData(header); |
| } |
| } else { |
| input.skipBlockletData(true); |
| } |
| scanMore = true; |
| } else { |
| isFinished = true; |
| scanMore = false; |
| } |
| } |
| } while (scanMore); |
| return hasNext; |
| } catch (FilterUnsupportedException e) { |
| throw new IOException("Failed to filter row in detail reader", e); |
| } |
| } |
| |
| @Override |
| public boolean nextKeyValue() throws IOException { |
| if (isFirstRow) { |
| isFirstRow = false; |
| initializeAtFirstRow(); |
| } |
| if (isFinished) { |
| return false; |
| } |
| |
| return nextRow(); |
| } |
| |
| @Override |
| public Void getCurrentKey() { |
| return null; |
| } |
| |
| @Override |
| public Object getCurrentValue() { |
| return outputValues; |
| } |
| |
| protected boolean isScanRequired(BlockletHeader header) { |
| if (filter != null && header.getBlocklet_index() != null) { |
| BlockletMinMaxIndex minMaxIndex = CarbonMetadataUtil |
| .convertExternalMinMaxIndex(header.getBlocklet_index().getMin_max_index()); |
| if (minMaxIndex != null) { |
| BitSet bitSet = filter |
| .isScanRequired(minMaxIndex.getMaxValues(), minMaxIndex.getMinValues(), |
| minMaxIndex.getIsMinMaxSet()); |
| return !bitSet.isEmpty(); |
| } |
| } |
| return true; |
| } |
| |
| protected void readRowFromStream() { |
| input.nextRow(); |
| short nullLen = input.readShort(); |
| BitSet nullBitSet = allNonNull; |
| if (nullLen > 0) { |
| nullBitSet = BitSet.valueOf(input.readBytes(nullLen)); |
| } |
| int colCount = 0; |
| // primitive type dimension |
| for (; colCount < isNoDictColumn.length; colCount++) { |
| if (nullBitSet.get(colCount)) { |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = null; |
| } |
| } else { |
| if (isNoDictColumn[colCount]) { |
| int v; |
| if (dimensionsIsVarcharTypeMap[colCount]) { |
| v = input.readInt(); |
| } else { |
| v = input.readShort(); |
| } |
| if (isRequired[colCount]) { |
| byte[] b = input.readBytes(v); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = b; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = DataTypeUtil |
| .getDataBasedOnDataTypeForNoDictionaryColumn(b, |
| storageColumns[colCount].getDataType()); |
| } |
| } else { |
| input.skipBytes(v); |
| } |
| } else if (null != directDictionaryGenerators[colCount]) { |
| if (isRequired[colCount]) { |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = input.copy(4); |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = |
| directDictionaryGenerators[colCount].getValueFromSurrogate(input.readInt()); |
| } else { |
| input.skipBytes(4); |
| } |
| } else { |
| input.skipBytes(4); |
| } |
| } else { |
| if (isRequired[colCount]) { |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = input.copy(4); |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = input.readInt(); |
| } else { |
| input.skipBytes(4); |
| } |
| } else { |
| input.skipBytes(4); |
| } |
| } |
| } |
| } |
| // complex type dimension |
| for (; colCount < dimensionCount; colCount++) { |
| if (nullBitSet.get(colCount)) { |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = null; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = null; |
| } |
| } else { |
| short v = input.readShort(); |
| if (isRequired[colCount]) { |
| byte[] b = input.readBytes(v); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = b; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = |
| queryTypes[colCount].getDataBasedOnDataType(ByteBuffer.wrap(b)); |
| } |
| } else { |
| input.skipBytes(v); |
| } |
| } |
| } |
| // measure |
| DataType dataType; |
| for (int msrCount = 0; msrCount < measureCount; msrCount++, colCount++) { |
| if (nullBitSet.get(colCount)) { |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = null; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = null; |
| } |
| } else { |
| dataType = measureDataTypes[msrCount]; |
| if (dataType == DataTypes.BOOLEAN) { |
| if (isRequired[colCount]) { |
| boolean v = input.readBoolean(); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = v; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = v; |
| } |
| } else { |
| input.skipBytes(1); |
| } |
| } else if (dataType == DataTypes.SHORT) { |
| if (isRequired[colCount]) { |
| short v = input.readShort(); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = v; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = v; |
| } |
| } else { |
| input.skipBytes(2); |
| } |
| } else if (dataType == DataTypes.INT) { |
| if (isRequired[colCount]) { |
| int v = input.readInt(); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = v; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = v; |
| } |
| } else { |
| input.skipBytes(4); |
| } |
| } else if (dataType == DataTypes.LONG) { |
| if (isRequired[colCount]) { |
| long v = input.readLong(); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = v; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = v; |
| } |
| } else { |
| input.skipBytes(8); |
| } |
| } else if (dataType == DataTypes.DOUBLE) { |
| if (isRequired[colCount]) { |
| double v = input.readDouble(); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = v; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = v; |
| } |
| } else { |
| input.skipBytes(8); |
| } |
| } else if (DataTypes.isDecimal(dataType)) { |
| int len = input.readShort(); |
| if (isRequired[colCount]) { |
| BigDecimal v = DataTypeUtil.byteToBigDecimal(input.readBytes(len)); |
| if (isFilterRequired[colCount]) { |
| filterValues[filterMap[colCount]] = v; |
| } |
| if (isProjectionRequired[colCount]) { |
| outputValues[projectionMap[colCount]] = |
| DataTypeUtil.getDataTypeConverter().convertFromBigDecimalToDecimal(v); |
| } |
| } else { |
| input.skipBytes(len); |
| } |
| } |
| } |
| } |
| } |
| |
| private void readRawRowFromStream() { |
| input.nextRow(); |
| short nullLen = input.readShort(); |
| BitSet nullBitSet = allNonNull; |
| if (nullLen > 0) { |
| nullBitSet = BitSet.valueOf(input.readBytes(nullLen)); |
| } |
| int colCount = 0; |
| // primitive type dimension |
| for (; colCount < isNoDictColumn.length; colCount++) { |
| if (nullBitSet.get(colCount)) { |
| outputValues[colCount] = CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY; |
| } else { |
| if (isNoDictColumn[colCount]) { |
| int v; |
| if (dimensionsIsVarcharTypeMap[colCount]) { |
| v = input.readInt(); |
| } else { |
| v = input.readShort(); |
| } |
| outputValues[colCount] = input.readBytes(v); |
| } else { |
| outputValues[colCount] = input.readInt(); |
| } |
| } |
| } |
| // complex type dimension |
| for (; colCount < dimensionCount; colCount++) { |
| if (nullBitSet.get(colCount)) { |
| outputValues[colCount] = null; |
| } else { |
| short v = input.readShort(); |
| outputValues[colCount] = input.readBytes(v); |
| } |
| } |
| // measure |
| DataType dataType; |
| for (int msrCount = 0; msrCount < measureCount; msrCount++, colCount++) { |
| if (nullBitSet.get(colCount)) { |
| outputValues[colCount] = null; |
| } else { |
| dataType = measureDataTypes[msrCount]; |
| if (dataType == DataTypes.BOOLEAN) { |
| outputValues[colCount] = input.readBoolean(); |
| } else if (dataType == DataTypes.SHORT) { |
| outputValues[colCount] = input.readShort(); |
| } else if (dataType == DataTypes.INT) { |
| outputValues[colCount] = input.readInt(); |
| } else if (dataType == DataTypes.LONG) { |
| outputValues[colCount] = input.readLong(); |
| } else if (dataType == DataTypes.DOUBLE) { |
| outputValues[colCount] = input.readDouble(); |
| } else if (DataTypes.isDecimal(dataType)) { |
| int len = input.readShort(); |
| outputValues[colCount] = DataTypeUtil.byteToBigDecimal(input.readBytes(len)); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public float getProgress() { |
| return 0; |
| } |
| |
| @Override |
| public void close() throws IOException { |
| if (null != input) { |
| input.close(); |
| } |
| } |
| } |
