private void updateAllocation ( ) { long totalAllocations = 0 ; for ( Long allocation : writerList . values ( ) ) { totalAllocations += allocation ; } if ( totalAllocations <= totalMemoryPool ) { scale = 1.0 ; } else { scale = ( double ) totalMemoryPool / totalAllocations ; LOG . warn ( String . format ( "Total allocation exceeds %.2f%% (%,d bytes) of heap memory\n" + "Scaling row group sizes to %.2f%% for %d writers" , 100 * memoryPoolRatio , totalMemoryPool , 100 * scale , writerList . size ( ) ) ) ; for ( Runnable callBack : callBacks . values ( ) ) { // we do not really want to start a new thread here. callBack . run ( ) ; } } int maxColCount = 0 ; for ( InternalParquetRecordWriter w : writerList . keySet ( ) ) { maxColCount = Math . max ( w . getSchema ( ) . getColumns ( ) . size ( ) , maxColCount ) ; } for ( Map . Entry < InternalParquetRecordWriter , Long > entry : writerList . entrySet ( ) ) { long newSize = ( long ) Math . floor ( entry . getValue ( ) * scale ) ; if ( scale < 1.0 && minMemoryAllocation > 0 && newSize < minMemoryAllocation ) { throw new ParquetRuntimeException ( String . format ( "New Memory allocation %d bytes" + " is smaller than the minimum allocation size of %d bytes." , newSize , minMemoryAllocation ) ) { } ; } entry . getKey ( ) . setRowGroupSizeThreshold ( newSize ) ; LOG . debug ( String . format ( "Adjust block size from %,d to %,d for writer: %s" , entry . getValue ( ) , newSize , entry . getKey ( ) ) ) ; } }
Update the allocated size of each writer based on the current allocations and pool size .