function readPipeline (readStream, opts, nosplit = false) {
  var beforeOffset = 0
  var beforeLimit = 0
  var offset = opts.offset || 0
  var limit = opts.limit
  var filter = massageFilters(opts.filter)
  return pump([
    readStream,
    nosplit ? undefined : split2(),
    through2.obj(function (row, enc, cb) {
      if (!row || !row.trim()) {
        // skip empty
        return cb()
      }

      // offset filter
      if (beforeOffset < offset) {
        beforeOffset++
        return cb()
      }

      // parse
      row = JSON.parse(row)

      // timestamp range filter
      var ts = (opts.since || opts.until) ? (new Date(row.timestamp)).getTime() : null
      if ('since' in opts && ts < opts.since) return cb()
      if ('until' in opts && ts > opts.until) return cb()

      // general string filters
      if (filter) {
        for (let k in filter) {
          if (!filter[k].includes(row[k])) return cb()
        }
      }

      // emit
      if (!limit || beforeLimit < limit) this.push(row)
      if (limit && ++beforeLimit === limit) readStream.destroy()
      cb()
    })
  ].filter(Boolean))
}
@param {any} readStream
@param {LogQueryOpts} opts
@returns {any}