Method getStream

c++/src/StripeStream.cc:82–126

  }

  std::unique_ptr<SeekableInputStream> StripeStreamsImpl::getStream(uint64_t columnId,
                                                                    proto::Stream_Kind kind,
                                                                    bool shouldStream) const {
    uint64_t offset = stripeStart_;
    uint64_t dataEnd =
        stripeInfo_.offset() + stripeInfo_.index_length() + stripeInfo_.data_length();
    MemoryPool* pool = reader_.getFileContents().pool;
    for (int i = 0; i < footer_.streams_size(); ++i) {
      const proto::Stream& stream = footer_.streams(i);
      if (stream.has_kind() && stream.kind() == kind &&
          stream.column() == static_cast<uint64_t>(columnId)) {
        uint64_t streamLength = stream.length();
        if (offset + streamLength > dataEnd) {
          std::stringstream msg;
          msg << "Malformed stream meta at stream index " << i << " in stripe " << stripeIndex_
              << ": streamOffset=" << offset << ", streamLength=" << streamLength
              << ", stripeOffset=" << stripeInfo_.offset()
              << ", stripeIndexLength=" << stripeInfo_.index_length()
              << ", stripeDataLength=" << stripeInfo_.data_length();
          throw ParseError(msg.str());
        }

        BufferSlice slice;
        if (readCache_) {
          ReadRange range{offset, streamLength};
          slice = readCache_->read(range);
        }

        uint64_t myBlock = shouldStream ? input_.getNaturalReadSize() : streamLength;
        std::unique_ptr<SeekableInputStream> seekableInput;
        if (slice.buffer) {
          seekableInput = std::make_unique<SeekableArrayInputStream>(
              slice.buffer->data() + slice.offset, slice.length);
        } else {
          seekableInput = std::make_unique<SeekableFileInputStream>(&input_, offset, streamLength,
                                                                    *pool, myBlock);
        }
        return createDecompressor(reader_.getCompression(), std::move(seekableInput),
                                  reader_.getCompressionSize(), *pool,
                                  reader_.getFileContents().readerMetrics);
      }
      offset += stream.length();
    }
    return nullptr;
  }

  MemoryPool& StripeStreamsImpl::getMemoryPool() const {
    return *reader_.getFileContents().pool;