Add ZSTD support for writing ORC and DWRF tables

[jainxrohit]

Add ZSTD support for writing ORC and DWRF tables
To enable ZSTD compression, use session property hive.compression_codec='ZSTD'.

== RELEASE NOTES ==

Hive Changes

• Add ZSTD support for writing ORC and DWRF tables. To enable ZSTD compression, use session property hive.compression_codec='ZSTD'.

[mbasmanova]

It would be very helpful to have an option of writing ZSTD-compressed ORC data. To move this forward, I suggest add fail-fast logic for queries that attempt to write ZSTD compressed Parquet files by introducing a check in HiveWritableTableHandle's constructor:

        if (!compressionCodec.isSupportedStorageFormat(tableStorageFormat)) {
            throw new PrestoException(GENERIC_USER_ERROR, String.format("%s compression is not supported with %s", compressionCodec.name(), tableStorageFormat.name()));
        }
        if (!compressionCodec.isSupportedStorageFormat(partitionStorageFormat)) {
            throw new PrestoException(GENERIC_USER_ERROR, String.format("%s compression is not supported with %s", compressionCodec.name(), partitionStorageFormat.name()));
        }

This check is supported by a new method in HiveCompressionCodec enum:

public enum HiveCompressionCodec
{
    NONE(null, CompressionKind.NONE, CompressionCodecName.UNCOMPRESSED, f -> true),
    SNAPPY(SnappyCodec.class, CompressionKind.SNAPPY, CompressionCodecName.SNAPPY, f -> true),
    GZIP(GzipCodec.class, CompressionKind.ZLIB, CompressionCodecName.GZIP, f -> true),
    ZSTD(null, CompressionKind.ZSTD, null, f -> f != HiveStorageFormat.PARQUET);

    private final Optional<Class<? extends CompressionCodec>> codec;
    private final CompressionKind orcCompressionKind;
    private final CompressionCodecName parquetCompressionCodec;
    private final Predicate<HiveStorageFormat> supportedStorageFormats;

    HiveCompressionCodec(
            Class<? extends CompressionCodec> codec,
            CompressionKind orcCompressionKind,
            CompressionCodecName parquetCompressionCodec,
            Predicate<HiveStorageFormat> supportedStorageFormats)
    {
        this.codec = Optional.ofNullable(codec);
        this.orcCompressionKind = requireNonNull(orcCompressionKind, "orcCompressionKind is null");
        this.parquetCompressionCodec = parquetCompressionCodec;
        this.supportedStorageFormats = requireNonNull(supportedStorageFormats, "supportedStorageFormats is null");
    }

    public Optional<Class<? extends CompressionCodec>> getCodec()
    {
        return codec;
    }

    public CompressionKind getOrcCompressionKind()
    {
        return orcCompressionKind;
    }

    public CompressionCodecName getParquetCompressionCodec()
    {
        return parquetCompressionCodec;
    }

    public boolean isSupportedStorageFormat(HiveStorageFormat storageFormat)
    {
        return supportedStorageFormats.test(storageFormat);
    }
}

TestHivePageSink would use the same check: if (codec == NONE || !codec.isSupportedStorageFormat(format)) {

Finally, I'd add ZSTD to OrcTester to make sure we get good test coverage.

@highker @wenleix @arhimondr @jainxrohit @bhhari Thoughts?

[mbasmanova]

@jainxrohit LGMT.

CC: @zhenxiao

[mbasmanova]

@jainxrohit There is a typo in commit message: prism.compression_codec -> hive.compression_codec . Same for PR description. Also, it would be nice to add a release note.

[jainxrohit]

@jainxrohit There is a typo in commit message: prism.compression_codec -> hive.compression_codec . Same for PR description. Also, it would be nice to add a release note.

@mbasmanova I will have to check this. In my local testing hive.compression_codec does not work, only prism.compression_codec works.

[zhenxiao]

looks good to me. We will add ZSTD for Parquet later :)

[jainxrohit]

@jainxrohit There is a typo in commit message: prism.compression_codec -> hive.compression_codec . Same for PR description. Also, it would be nice to add a release note.

@mbasmanova I will have to check this. In my local testing hive.compression_codec does not work, only prism.compression_codec works.

I have fixed it.

[mbasmanova]

@jainxrohit Commit message contains some duplicate sentences. Perhaps, update to something like:

Add ZSTD support for writing ORC and DWRF tables

To enable ZSTD compression, use session property hive.compression_codec='ZSTD'.

[wenleix]

nit: what about format -> true ? ditto for other lambdas.

[mbasmanova]

@wenleix Wenlei, do you have any further comments? Otherwise, I'll merge.