code: func newChannelOfValuesWithSameKey(name string, sortedChan io.Reader, indexes []int) chan util.Row {
  	writer := make(chan util.Row, 1024)
  	go func() {
  
  		defer close(writer)
  
  		firstRow, err := util.ReadRow(sortedChan)
  		if err != nil {
  			if err != io.EOF {
  				fmt.Fprintf(os.Stderr, "%s join read first row error: %v\n", name, err)
  			}
  			return
  		}
  		// fmt.Printf("%s join read len=%d, row: %s\n", name, len(row), row[0])
  
  		firstRow.UseKeys(indexes)
  
  		row := util.Row{
  			T: firstRow.T,
  			K: firstRow.K,
  			V: []interface{}{firstRow.V},
  		}
  
  		for {
  			newRow, err := util.ReadRow(sortedChan)
  			if err != nil {
  				if err != io.EOF {
  					fmt.Fprintf(os.Stderr, "join read row error: %v", err)
  				}
  				break
  			}
  			newRow.UseKeys(indexes)
  			x := util.Compare(row.K, newRow.K)
  			if x == 0 {
  				row.V = append(row.V, newRow.V)
  			} else {
  				writer <- row
  				row.K = newRow.K
  				row.V = []interface{}{newRow.V}
  			}
  			row.T = max(row.T, newRow.T)
  		}
  		writer <- row
  	}()
  
  	return writer
  }
docs: // create a channel to aggregate values of the same key
  // automatically close original sorted channel