@Override
    public OutputReceiver createDataStream(StreamName name) throws IOException {
      OutputReceiver or = streams.get(name);
      if (or != null) return or;
      if (isNeeded(name)) {
        if (LlapIoImpl.LOG.isTraceEnabled()) {
          LlapIoImpl.LOG.trace("Creating cache receiver for " + name);
        }
        CacheOutputReceiver cor = new CacheOutputReceiver(bufferManager, name, bufferFactory, isStopped);
        or = cor;
        List<CacheOutputReceiver> list = colStreams.get(name.getColumn());
        if (list == null) {
          list = new ArrayList<>();
          colStreams.put(name.getColumn(), list);
        }
        list.add(cor);
      } else {
        if (LlapIoImpl.LOG.isTraceEnabled()) {
          LlapIoImpl.LOG.trace("Creating null receiver for " + name);
        }
        or = new NullOutputReceiver(name);
      }
      streams.put(name, or);
      return or;
    }
For !doesSourceHaveIncludes case, stores global column IDs to verify writer columns.
For doesSourceHaveIncludes case, stores source column IDs used to map things.
llap-server/src/java/org/apache/hadoop/hive/llap/io/encoded/SerDeEncodedDataReader.java