void readBatch ( int total , WritableColumnVector column ) throws IOException { int rowId = 0 ; WritableColumnVector dictionaryIds = null ; if ( dictionary != null ) { // SPARK-16334: We only maintain a single dictionary per row batch, so that it can be used to // decode all previous dictionary encoded pages if we ever encounter a non-dictionary encoded // page. dictionaryIds = column . reserveDictionaryIds ( total ) ; } while ( total > 0 ) { // Compute the number of values we want to read in this page. int leftInPage = ( int ) ( endOfPageValueCount - valuesRead ) ; if ( leftInPage == 0 ) { readPage ( ) ; leftInPage = ( int ) ( endOfPageValueCount - valuesRead ) ; } int num = Math . min ( total , leftInPage ) ; PrimitiveType . PrimitiveTypeName typeName = descriptor . getPrimitiveType ( ) . getPrimitiveTypeName ( ) ; if ( isCurrentPageDictionaryEncoded ) { // Read and decode dictionary ids. defColumn . readIntegers ( num , dictionaryIds , column , rowId , maxDefLevel , ( VectorizedValuesReader ) dataColumn ) ; // TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we need to post process // the values to add microseconds precision. if ( column . hasDictionary ( ) || ( rowId == 0 && ( typeName == PrimitiveType . PrimitiveTypeName . INT32 || ( typeName == PrimitiveType . PrimitiveTypeName . INT64 && originalType != OriginalType . TIMESTAMP_MILLIS ) || typeName == PrimitiveType . PrimitiveTypeName . FLOAT || typeName == PrimitiveType . PrimitiveTypeName . DOUBLE || typeName == PrimitiveType . PrimitiveTypeName . BINARY ) ) ) { // Column vector supports lazy decoding of dictionary values so just set the dictionary. // We can't do this if rowId != 0 AND the column doesn't have a dictionary (i.e. some // non-dictionary encoded values have already been added). column . setDictionary ( new ParquetDictionary ( dictionary ) ) ; } else { decodeDictionaryIds ( rowId , num , column , dictionaryIds ) ; } } else { if ( column . hasDictionary ( ) && rowId != 0 ) { // This batch already has dictionary encoded values but this new page is not. The batch // does not support a mix of dictionary and not so we will decode the dictionary. decodeDictionaryIds ( 0 , rowId , column , column . getDictionaryIds ( ) ) ; } column . setDictionary ( null ) ; switch ( typeName ) { case BOOLEAN : readBooleanBatch ( rowId , num , column ) ; break ; case INT32 : readIntBatch ( rowId , num , column ) ; break ; case INT64 : readLongBatch ( rowId , num , column ) ; break ; case INT96 : readBinaryBatch ( rowId , num , column ) ; break ; case FLOAT : readFloatBatch ( rowId , num , column ) ; break ; case DOUBLE : readDoubleBatch ( rowId , num , column ) ; break ; case BINARY : readBinaryBatch ( rowId , num , column ) ; break ; case FIXED_LEN_BYTE_ARRAY : readFixedLenByteArrayBatch ( rowId , num , column , descriptor . getPrimitiveType ( ) . getTypeLength ( ) ) ; break ; default : throw new IOException ( "Unsupported type: " + typeName ) ; } } valuesRead += num ; rowId += num ; total -= num ; } }
Reads total values from this columnReader into column .