presto-orc/src/main/java/io/prestosql/orc/stream/ValueStreams.java/ValueStreams/createValueStreams
class ValueStreams:    public static ValueInputStream<?> createValueStreams(
            StreamId streamId,
            OrcChunkLoader chunkLoader,
            OrcTypeKind type,
            ColumnEncodingKind encoding)
    {
        if (streamId.getStreamKind() == PRESENT) {
            return new BooleanInputStream(new OrcInputStream(chunkLoader));
        }

        // dictionary length and data streams are unsigned int streams
        if ((encoding == DICTIONARY || encoding == DICTIONARY_V2) && (streamId.getStreamKind() == LENGTH || streamId.getStreamKind() == DATA)) {
            return createLongStream(new OrcInputStream(chunkLoader), encoding, false);
        }

        if (streamId.getStreamKind() == DATA) {
            switch (type) {
                case BOOLEAN:
                    return new BooleanInputStream(new OrcInputStream(chunkLoader));
                case BYTE:
                    return new ByteInputStream(new OrcInputStream(chunkLoader));
                case SHORT:
                case INT:
                case LONG:
                case DATE:
                    return createLongStream(new OrcInputStream(chunkLoader), encoding, true);
                case FLOAT:
                    return new FloatInputStream(new OrcInputStream(chunkLoader));
                case DOUBLE:
                    return new DoubleInputStream(new OrcInputStream(chunkLoader));
                case STRING:
                case VARCHAR:
                case CHAR:
                case BINARY:
                    return new ByteArrayInputStream(new OrcInputStream(chunkLoader));
                case TIMESTAMP:
                    return createLongStream(new OrcInputStream(chunkLoader), encoding, true);
                case DECIMAL:
                    return new DecimalInputStream(chunkLoader);
            }
        }

        // length stream of a direct encoded string or binary column
        if (streamId.getStreamKind() == LENGTH) {
            switch (type) {
                case STRING:
                case VARCHAR:
                case CHAR:
                case BINARY:
                case MAP:
                case LIST:
                    return createLongStream(new OrcInputStream(chunkLoader), encoding, false);
            }
        }

        // length (nanos) of a timestamp column
        if (type == TIMESTAMP && streamId.getStreamKind() == SECONDARY) {
            return createLongStream(new OrcInputStream(chunkLoader), encoding, false);
        }

        // scale of a decimal column
        if (type == DECIMAL && streamId.getStreamKind() == SECONDARY) {
            // specification (https://orc.apache.org/docs/encodings.html) says scale stream is unsigned,
            // however Hive writer stores scale as signed integer (org.apache.hadoop.hive.ql.io.orc.WriterImpl.DecimalTreeWriter)
            // BUG link: https://issues.apache.org/jira/browse/HIVE-13229
            return createLongStream(new OrcInputStream(chunkLoader), encoding, true);
        }

        if (streamId.getStreamKind() == DICTIONARY_DATA) {
            switch (type) {
                case STRING:
                case VARCHAR:
                case CHAR:
                case BINARY:
                    return new ByteArrayInputStream(new OrcInputStream(chunkLoader));
            }
        }

        throw new IllegalArgumentException(format("Unsupported column type %s for stream %s with encoding %s", type, streamId, encoding));
    }