code: func ProcessRow(reader io.Reader, indexes []int, f func(*Row) error) (err error) {
  	return ProcessMessage(reader, func(input []byte) error {
  		// read the row
  		row, err := DecodeRow(input)
  		if err != nil {
  			return fmt.Errorf("DoLocalDistinct error %v: %+v", err, input)
  		}
  		row.UseKeys(indexes)
  		return f(row)
  	})
  }
docs: // ProcessRow Reads and processes rows until EOF