func()

in banyand/tsdb/seriesdb.go [483:538]

• 56 lines of code
• 15 McCabe index (conditional complexity)

func (s *seriesDB) Search(ctx context.Context, path Path, filter index.Filter, order *OrderBy) (SeriesList, error) {
	if s.invertedIndex == nil || s.lsmIndex == nil {
		return nil, errors.New("search is not supported")
	}
	if path.isFull {
		return s.List(ctx, path)
	}
	if order == nil {
		return s.filterSeries(ctx, path, filter)
	}
	fieldKey := index.FieldKey{
		IndexRuleID: order.Index.GetMetadata().Id,
	}
	var iter index.FieldIterator
	var err error
	switch order.Index.Type {
	case databasev1.IndexRule_TYPE_TREE:
		iter, err = s.lsmIndex.Iterator(fieldKey, rangeOpts, order.Sort)
	case databasev1.IndexRule_TYPE_INVERTED:
		iter, err = s.invertedIndex.Iterator(fieldKey, rangeOpts, order.Sort)
	default:
		return nil, errUnspecifiedIndexType
	}
	if err != nil {
		return nil, err
	}
	defer func() {
		err = multierr.Append(err, iter.Close())
	}()
	var pl posting.List
	if pl, err = s.seriesFilter(ctx, path, filter); err != nil {
		return nil, err
	}
	seriesList := make([]Series, 0)
	for iter.Next() {
		pv := iter.Val().Value
		if err = pv.Intersect(pl); err != nil {
			return nil, err
		}
		if pv.IsEmpty() {
			continue
		}
		pIter := pv.Iterator()
		for pIter.Next() {
			var series Series
			if series, err = s.GetByID(common.SeriesID(pIter.Current())); err != nil {
				return nil, multierr.Append(err, pIter.Close())
			}
			seriesList = append(seriesList, series)
		}
		if err = pIter.Close(); err != nil {
			return nil, err
		}
	}
	return seriesList, err
}