presto-orc/src/main/java/io/prestosql/orc/DictionaryCompressionOptimizer.java/DictionaryCompressionOptimizer/optimize
class DictionaryCompressionOptimizer:    public void optimize(int bufferedBytes, int stripeRowCount)
    {
        // recompute the dictionary memory usage
        dictionaryMemoryBytes = allWriters.stream()
                .filter(writer -> !writer.isDirectEncoded())
                .mapToInt(DictionaryColumnManager::getDictionaryBytes)
                .sum();

        // update the dictionary growth history
        allWriters.stream()
                .filter(writer -> !writer.isDirectEncoded())
                .forEach(column -> column.updateHistory(stripeRowCount));

        if (dictionaryMemoryBytes <= dictionaryMemoryMaxBytesLow) {
            return;
        }

        // before any further checks, convert all low compression streams
        bufferedBytes = convertLowCompressionStreams(bufferedBytes);

        if (dictionaryMemoryBytes <= dictionaryMemoryMaxBytesLow || bufferedBytes >= stripeMaxBytes) {
            return;
        }

        // calculate size of non-dictionary columns by removing the buffered size of dictionary columns
        int nonDictionaryBufferedBytes = bufferedBytes;
        for (DictionaryColumnManager dictionaryWriter : allWriters) {
            if (!dictionaryWriter.isDirectEncoded()) {
                nonDictionaryBufferedBytes -= dictionaryWriter.getBufferedBytes();
            }
        }

        // convert dictionary columns to direct until we are below the high memory limit
        while (!directConversionCandidates.isEmpty() && dictionaryMemoryBytes > dictionaryMemoryMaxBytesHigh && bufferedBytes < stripeMaxBytes) {
            DictionaryCompressionProjection projection = selectDictionaryColumnToConvert(nonDictionaryBufferedBytes, stripeRowCount);
            int selectDictionaryColumnBufferedBytes = toIntExact(projection.getColumnToConvert().getBufferedBytes());

            OptionalInt directBytes = tryConvertToDirect(projection.getColumnToConvert(), getMaxDirectBytes(bufferedBytes));
            if (directBytes.isPresent()) {
                bufferedBytes = bufferedBytes + directBytes.getAsInt() - selectDictionaryColumnBufferedBytes;
                nonDictionaryBufferedBytes += directBytes.getAsInt();
            }
        }

        if (bufferedBytes >= stripeMaxBytes) {
            return;
        }

        // if the stripe is larger then the minimum stripe size, we are not required to convert any more dictionary columns to direct
        if (bufferedBytes >= stripeMinBytes) {
            // check if we can get better compression by converting a dictionary column to direct.  This can happen when then there are multiple
            // dictionary columns and one does not compress well, so if we convert it to direct we can continue to use the existing dictionaries
            // for the other columns.
            double currentCompressionRatio = currentCompressionRatio(nonDictionaryBufferedBytes);
            while (!directConversionCandidates.isEmpty() && bufferedBytes < stripeMaxBytes) {
                DictionaryCompressionProjection projection = selectDictionaryColumnToConvert(nonDictionaryBufferedBytes, stripeRowCount);
                if (projection.getPredictedFileCompressionRatio() < currentCompressionRatio) {
                    return;
                }

                int selectDictionaryColumnBufferedBytes = toIntExact(projection.getColumnToConvert().getBufferedBytes());
                OptionalInt directBytes = tryConvertToDirect(projection.getColumnToConvert(), getMaxDirectBytes(bufferedBytes));
                if (directBytes.isPresent()) {
                    bufferedBytes = bufferedBytes + directBytes.getAsInt() - selectDictionaryColumnBufferedBytes;
                    nonDictionaryBufferedBytes += directBytes.getAsInt();
                }
            }
        }
    }