code: func (sink *influxdbSink) composeAggregateQuery(metricName string, labels map[string]string, aggregations []core.AggregationType, metricKeys []core.HistoricalKey, start, end time.Time, bucketSize time.Duration) string {
  	seriesName, fieldName := sink.metricToSeriesAndField(metricName)
  
  	var bucketSizeNanoSeconds int64 = 0
  	if bucketSize != 0 {
  		bucketSizeNanoSeconds = int64(bucketSize.Nanoseconds() / int64(time.Microsecond/time.Nanosecond))
  	}
  
  	queries := make([]string, len(metricKeys))
  	for i, key := range metricKeys {
  		pred := sink.keyToSelector(key)
  		if labels != nil {
  			pred += fmt.Sprintf(" AND %s", sink.labelsToPredicate(labels))
  		}
  		if !start.IsZero() {
  			pred += fmt.Sprintf(" AND time > '%s'", start.Format(time.RFC3339))
  		}
  		if !end.IsZero() {
  			pred += fmt.Sprintf(" AND time < '%s'", end.Format(time.RFC3339))
  		}
  
  		aggParts := make([]string, len(aggregations))
  		for i, agg := range aggregations {
  			aggParts[i] = sink.aggregationFunc(agg, fieldName)
  		}
  
  		queries[i] = fmt.Sprintf("SELECT %s FROM %q WHERE %s", strings.Join(aggParts, ", "), seriesName, pred)
  
  		if bucketSize != 0 {
  			// group by time requires we have at least one time bound
  			if start.IsZero() && end.IsZero() {
  				queries[i] += fmt.Sprintf(" AND time < now()")
  			}
  
  			// fill(none) makes sure we skip data points will null values (otherwise we'll get a *bunch* of null
  			// values when we go back beyond the time where we started collecting data).
  			queries[i] += fmt.Sprintf(" GROUP BY time(%vu) fill(none)", bucketSizeNanoSeconds)
  		}
  	}
  
  	return strings.Join(queries, "; ")
  }
docs: // composeAggregateQuery creates the InfluxQL query to fetch the given aggregation values