presto-rcfile/src/main/java/io/prestosql/rcfile/RcFileReader.java/RcFileReader/RcFileReader
class RcFileReader:    private RcFileReader(
            RcFileDataSource dataSource,
            RcFileEncoding encoding,
            Map<Integer, Type> readColumns,
            RcFileCodecFactory codecFactory,
            long offset,
            long length,
            DataSize bufferSize,
            Optional<RcFileWriteValidation> writeValidation)
            throws IOException
    {
        this.dataSource = requireNonNull(dataSource, "rcFileDataSource is null");
        this.readColumns = ImmutableMap.copyOf(requireNonNull(readColumns, "readColumns is null"));
        this.input = new ChunkedSliceInput(new DataSourceSliceLoader(dataSource), toIntExact(bufferSize.toBytes()));

        this.writeValidation = requireNonNull(writeValidation, "writeValidation is null");
        this.writeChecksumBuilder = writeValidation.map(validation -> createWriteChecksumBuilder(readColumns));

        verify(offset >= 0, "offset is negative");
        verify(offset < dataSource.getSize(), "offset is greater than data size");
        verify(length >= 1, "length must be at least 1");
        this.length = length;
        this.end = offset + length;
        verify(end <= dataSource.getSize(), "offset plus length is greater than data size");

        // read header
        Slice magic = input.readSlice(RCFILE_MAGIC.length());
        boolean compressed;
        if (RCFILE_MAGIC.equals(magic)) {
            version = input.readByte();
            verify(version <= CURRENT_VERSION, "RCFile version %s not supported: %s", version, dataSource);
            validateWrite(validation -> validation.getVersion() == version, "Unexpected file version");
            compressed = input.readBoolean();
        }
        else if (SEQUENCE_FILE_MAGIC.equals(magic)) {
            validateWrite(validation -> false, "Expected file to start with RCFile magic");

            // first version of RCFile used magic SEQ with version 6
            byte sequenceFileVersion = input.readByte();
            verify(sequenceFileVersion == SEQUENCE_FILE_VERSION, "File %s is a SequenceFile not an RCFile", dataSource);

            // this is the first version of RCFile
            this.version = FIRST_VERSION;

            Slice keyClassName = readLengthPrefixedString(input);
            Slice valueClassName = readLengthPrefixedString(input);
            verify(RCFILE_KEY_BUFFER_NAME.equals(keyClassName) && RCFILE_VALUE_BUFFER_NAME.equals(valueClassName), "File %s is a SequenceFile not an RCFile", dataSource);
            compressed = input.readBoolean();

            // RC file is never block compressed
            if (input.readBoolean()) {
                throw corrupt("File %s is a SequenceFile not an RCFile", dataSource);
            }
        }
        else {
            throw corrupt("File %s is not an RCFile", dataSource);
        }

        // setup the compression codec
        if (compressed) {
            String codecClassName = readLengthPrefixedString(input).toStringUtf8();
            validateWrite(validation -> validation.getCodecClassName().equals(Optional.of(codecClassName)), "Unexpected compression codec");
            this.decompressor = codecFactory.createDecompressor(codecClassName);
        }
        else {
            validateWrite(validation -> validation.getCodecClassName().equals(Optional.empty()), "Expected file to be compressed");
            this.decompressor = null;
        }

        // read metadata
        int metadataEntries = Integer.reverseBytes(input.readInt());
        verify(metadataEntries >= 0, "Invalid metadata entry count %s in RCFile %s", metadataEntries, dataSource);
        verify(metadataEntries <= MAX_METADATA_ENTRIES, "Too many metadata entries (%s) in RCFile %s", metadataEntries, dataSource);
        ImmutableMap.Builder<String, String> metadataBuilder = ImmutableMap.builder();
        for (int i = 0; i < metadataEntries; i++) {
            metadataBuilder.put(readLengthPrefixedString(input).toStringUtf8(), readLengthPrefixedString(input).toStringUtf8());
        }
        metadata = metadataBuilder.build();
        validateWrite(validation -> validation.getMetadata().equals(metadata), "Unexpected metadata");

        // get column count from metadata
        String columnCountString = metadata.get(COLUMN_COUNT_METADATA_KEY);
        try {
            columnCount = Integer.parseInt(columnCountString);
        }
        catch (NumberFormatException e) {
            throw corrupt("Invalid column count %s in RCFile %s", columnCountString, dataSource);
        }

        // initialize columns
        verify(columnCount <= MAX_COLUMN_COUNT, "Too many columns (%s) in RCFile %s", columnCountString, dataSource);
        columns = new Column[columnCount];
        for (Entry<Integer, Type> entry : readColumns.entrySet()) {
            if (entry.getKey() < columnCount) {
                ColumnEncoding columnEncoding = encoding.getEncoding(entry.getValue());
                columns[entry.getKey()] = new Column(columnEncoding, decompressor);
            }
        }

        // read sync bytes
        syncFirst = input.readLong();
        validateWrite(validation -> validation.getSyncFirst() == syncFirst, "Unexpected sync sequence");
        syncSecond = input.readLong();
        validateWrite(validation -> validation.getSyncSecond() == syncSecond, "Unexpected sync sequence");

        // seek to first sync point within the specified region, unless the region starts at the beginning
        // of the file.  In that case, the reader owns all row groups up to the first sync point.
        if (offset != 0) {
            // if the specified file region does not contain the start of a sync sequence, this call will close the reader
            seekToFirstRowGroupInRange(offset, length);
        }
    }