parquet-hadoop/src/main/java/org/apache/parquet/hadoop/MemoryManager.java
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()));
    }
  }
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()));
    }
  }
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.
Update the allocated size of each writer based on the current allocations and pool size .