Reduce memory utilization on the Driver during the commit phase

[arhimondr]

In Presto each writer produces a PartitionUpdate object that contains file names and other meta information for the files being written. This information is then collected on the driver to perform a final commit (do file renames). In some cases this meta information could be quite large. This patch tries to optimize several things:

• Reduce PartitionUpdate memory footprint on the Driver by serializing to SMILE instead of JSON and applying ZSTD compression
• Release serializes and compressed pages as soon as they read by the engine on the driver. This should help avoid double memory utilization

== RELEASE NOTES ==

Presto on Spark Changes
* Reduce commit memory footprint on the Driver

[arhimondr]

This is all in addition to the #16036, that should significantly reduce memory utilization on the driver as the statistic pages no longer have to be buffered in the TableFinishOperator for the Presto on Spark usecase (Thanks @viczhang861 for optimizing it!)

[viczhang861]

Try to make title "Release inmemory input pages incrementally" better, inmemory is only used for PrestoSparkTaskInputs

[viczhang861]

extra "the" after "compress"

[tdcmeehan]

Suggested change

	try (ByteArrayOutputStream output = new ByteArrayOutputStream();
	ZstdOutputStreamNoFinalizer zstdOutput = new ZstdOutputStreamNoFinalizer(output)) {
	codec.writeBytes(zstdOutput, instance);
	zstdOutput.close();
	output.close();
	return output.toByteArray();
	}
	catch (IOException e) {
	throw new UncheckedIOException(e);
	}
	}
	try (ByteArrayOutputStream output = new ByteArrayOutputStream()) {
	try (ZstdOutputStreamNoFinalizer zstdOutput = new ZstdOutputStreamNoFinalizer(output)) {
	codec.writeBytes(zstdOutput, instance);
	}
	return output.toByteArray();
	}
	catch (IOException e) {
	throw new UncheckedIOException(e);
	}
	}

[arhimondr]

In theory it should be the same. Java guarantees to close all resources in a reverse order.

public class Main
{
    private static class Closeable1
            implements Closeable
    {

        @Override
        public void close()
        {
            System.out.println("Close Closeable1");
        }
    }

    private static class Closeable2
            implements Closeable
    {

        @Override
        public void close()
        {
            System.out.println("Close Closeable2");
        }
    }

    public static void main(String[] args)
    {
        try (Closeable1 closeable1 = new Closeable1(); Closeable2 closeable2 = new Closeable2()) {
            System.out.println("Body");
        }
    }
}

Prints

Body
Close Closeable2
Close Closeable1

[tdcmeehan]

The comment was more around not calling close() explicitly, since the try-with-resources does that.

[arhimondr]

Oh, sorry. I misunderstood. Let me create a patch

[arhimondr]

Actually I'm not sure if it is correct to call return output.toByteArray() before the ByteArrayOutputStream is closed? The implementation allows it, but I wonder if that's what is expected?

[tdcmeehan]

I guess you can't just call remove() on the iterator because that's a code change in Spark?

[arhimondr]

In theory remove on the iterator for an ArrayList is an O(N) operation (because it has to shift the "tail"). Although in practice I don't think it is going to be an issue, just being on a safer side (just in case there's a query that produces a list with a number of pages that would make this complexity to create a problem)