@Override
    public Footer readFooter(HiveWriterVersion hiveWriterVersion,
            InputStream inputStream,
            DwrfEncryptionProvider dwrfEncryptionProvider,
            DwrfKeyProvider dwrfKeyProvider,
            OrcDataSource orcDataSource,
            Optional<OrcDecompressor> decompressor)
            throws IOException
    {
        long cpuStart = THREAD_MX_BEAN.getCurrentThreadCpuTime();
        CodedInputStream input = CodedInputStream.newInstance(inputStream);
        DwrfProto.Footer footer = DwrfProto.Footer.parseFrom(input);
        List<ColumnStatistics> fileStats = toColumnStatistics(hiveWriterVersion, footer.getStatisticsList(), false);
        List<StripeInformation> fileStripes = toStripeInformation(footer.getStripesList());
        List<OrcType> types = toType(footer.getTypesList());
        Optional<DwrfEncryption> encryption = footer.hasEncryption() ? Optional.of(toEncryption(footer.getEncryption())) : Optional.empty();
        Optional<List<Integer>> stripeCacheOffsets = Optional.of(footer.getStripeCacheOffsetsList());

        if (encryption.isPresent()) {
            Map<Integer, Slice> keys = dwrfKeyProvider.getIntermediateKeys(types);
            EncryptionLibrary encryptionLibrary = dwrfEncryptionProvider.getEncryptionLibrary(encryption.get().getKeyProvider());
            fileStats = decryptAndCombineFileStatistics(hiveWriterVersion, encryption.get(), encryptionLibrary, fileStats, fileStripes, keys, orcDataSource, decompressor);
        }
        runtimeStats.addMetricValue("DwrfReadFooterTimeNanos", RuntimeUnit.NANO, THREAD_MX_BEAN.getCurrentThreadCpuTime() - cpuStart);

        OptionalLong rawSize = footer.hasRawDataSize() ? OptionalLong.of(footer.getRawDataSize()) : OptionalLong.empty();
        return new Footer(
                footer.getNumberOfRows(),
                footer.getRowIndexStride(),
                rawSize,
                fileStripes,
                types,
                fileStats,
                toUserMetadata(footer.getMetadataList()),
                encryption,
                stripeCacheOffsets);
    }
@Test
    public void testReadFooter()
            throws IOException
    {
        long numberOfRows = 10;
        int rowIndexStride = 11;
        List<Integer> stripeCacheOffsets = ImmutableList.of(1, 2, 3);

        for (OptionalLong rawDataSize : ImmutableList.of(OptionalLong.empty(), OptionalLong.of(1_000_123))) {
            DwrfProto.Footer.Builder protoFooterBuilder = DwrfProto.Footer.newBuilder()
                    .setNumberOfRows(numberOfRows)
                    .setRowIndexStride(rowIndexStride)
                    .addAllStripeCacheOffsets(stripeCacheOffsets);

            if (rawDataSize.isPresent()) {
                protoFooterBuilder.setRawDataSize(rawDataSize.getAsLong());
            }

            DwrfProto.Footer protoFooter = protoFooterBuilder.build();
            Footer footer = convertToFooter(protoFooter);

            assertEquals(footer.getNumberOfRows(), numberOfRows);
            assertEquals(footer.getRowsInRowGroup(), rowIndexStride);
            assertEquals(footer.getDwrfStripeCacheOffsets().get(), stripeCacheOffsets);
            assertEquals(footer.getRawSize(), rawDataSize);
            assertEquals(footer.getStripes(), Collections.emptyList());
        }
    }