private static TreeReader createEncodedTreeReader(TypeDescription schema, List<OrcProto.ColumnEncoding> encodings,
      OrcEncodedColumnBatch batch, CompressionCodec codec, Context context, final boolean useDecimal64ColumnVectors)
      throws IOException {
    int columnIndex = schema.getId();
    ColumnStreamData[] streamBuffers = null;
    List<ColumnVector> vectors = null;
    if (batch.hasData(columnIndex)) {
      streamBuffers = batch.getColumnData(columnIndex);
    } else if (batch.hasVectors(columnIndex)) {
      vectors = batch.getColumnVectors(columnIndex);
    } else {
      // A struct column can have a null child column
      LOG.debug("Batch has no data for {}: {}", columnIndex, batch);
      return null;
    }

    // EncodedColumnBatch is already decompressed, we don't really need to pass codec.
    // But we need to know if the original data is compressed or not. This is used to skip
    // positions in row index properly. If the file is originally compressed,
    // then 1st position (compressed offset) in row index should be skipped to get
    // uncompressed offset, else 1st position should not be skipped.
    // TODO: there should be a better way to do this, code just needs to be modified
    OrcProto.ColumnEncoding columnEncoding = encodings.get(columnIndex);

    // stream buffers are arranged in enum order of stream kind
    ColumnStreamData present = null, data = null, dictionary = null, lengths = null, secondary = null;
    if (streamBuffers != null) {
      present = streamBuffers[OrcProto.Stream.Kind.PRESENT_VALUE];
      data = streamBuffers[OrcProto.Stream.Kind.DATA_VALUE];
      dictionary = streamBuffers[OrcProto.Stream.Kind.DICTIONARY_DATA_VALUE];
      lengths = streamBuffers[OrcProto.Stream.Kind.LENGTH_VALUE];
      secondary = streamBuffers[OrcProto.Stream.Kind.SECONDARY_VALUE];
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug(
          "columnIndex: {} columnType: {} streamBuffers.length: {} vectors: {} columnEncoding: {}" + " present: {} data: {} dictionary: {} lengths: {} secondary: {} tz: {}",
          columnIndex, schema, streamBuffers == null ? 0 : streamBuffers.length, vectors == null ? 0 : vectors.size(),
          columnEncoding, present != null, data, dictionary != null, lengths != null, secondary != null,
          context.getWriterTimezone());
    }
    // TODO: get rid of the builders - they serve no purpose... just call ctors directly.
    switch (schema.getCategory()) {
    case BINARY:
    case BOOLEAN:
    case BYTE:
    case SHORT:
    case INT:
    case LONG:
    case FLOAT:
    case DOUBLE:
    case CHAR:
    case VARCHAR:
    case STRING:
    case DECIMAL:
    case TIMESTAMP:
    case DATE:
      return getPrimitiveTreeReader(columnIndex, schema, codec, columnEncoding, present, data, dictionary, lengths,
          secondary, context, vectors, useDecimal64ColumnVectors);
    case LIST:
      assert vectors == null; // Not currently supported.
      TypeDescription elementType = schema.getChildren().get(0);
      TreeReader elementReader = createEncodedTreeReader(elementType, encodings, batch, codec, context,
          useDecimal64ColumnVectors);
      return ListStreamReader.builder().setColumnIndex(columnIndex).setColumnEncoding(columnEncoding)
          .setCompressionCodec(codec).setPresentStream(present).setLengthStream(lengths).setElementReader(elementReader)
          .setContext(context).build();
    case MAP:
      assert vectors == null; // Not currently supported.
      TypeDescription keyType = schema.getChildren().get(0);
      TypeDescription valueType = schema.getChildren().get(1);
      TreeReader keyReader = createEncodedTreeReader(keyType, encodings, batch, codec, context,
          useDecimal64ColumnVectors);
      TreeReader valueReader = createEncodedTreeReader(valueType, encodings, batch, codec, context,
          useDecimal64ColumnVectors);
      return MapStreamReader.builder().setColumnIndex(columnIndex).setColumnEncoding(columnEncoding)
          .setCompressionCodec(codec).setPresentStream(present).setLengthStream(lengths).setKeyReader(keyReader)
          .setValueReader(valueReader).setContext(context).build();
    case STRUCT: {
      assert vectors == null; // Not currently supported.
      int childCount = schema.getChildren().size();
      TreeReader[] childReaders = new TreeReader[childCount];
      for (int i = 0; i < childCount; i++) {
        TypeDescription childType = schema.getChildren().get(i);
        childReaders[i] = createEncodedTreeReader(childType, encodings, batch, codec, context,
            useDecimal64ColumnVectors);
      }
      return StructStreamReader.builder().setColumnIndex(columnIndex).setCompressionCodec(codec)
          .setColumnEncoding(columnEncoding).setPresentStream(present).setChildReaders(childReaders).setContext(context)
          .build();
    }
    case UNION: {
      assert vectors == null; // Not currently supported.
      int childCount = schema.getChildren().size();
      TreeReader[] childReaders = new TreeReader[childCount];
      for (int i = 0; i < childCount; i++) {
        TypeDescription childType = schema.getChildren().get(i);
        childReaders[i] = createEncodedTreeReader(childType, encodings, batch, codec, context,
            useDecimal64ColumnVectors);
      }
      return UnionStreamReader.builder().setColumnIndex(columnIndex).setCompressionCodec(codec)
          .setColumnEncoding(columnEncoding).setPresentStream(present).setDataStream(data).setChildReaders(childReaders)
          .setContext(context).build();
    }
    default:
      throw new UnsupportedOperationException("Data type not supported: " + schema);
    }
  }
We choose to use a toy programming language, so we cannot use multiple inheritance.
If we could, we could have this inherit TreeReader to contain the common impl, and then
have e.g. SettableIntTreeReader inherit both Settable... and Int.. TreeReader-s.
Instead, we have a settable interface that the caller will cast to and call setBuffers.
ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedTreeReaderFactory.java