// buildBucketFeedback build the feedback for each bucket from the histogram feedback.
513186
func buildBucketFeedback(h *Histogram, feedback *QueryFeedback) (map[int]*BucketFeedback, int) {
	bktID2FB := make(map[int]*BucketFeedback)
	if len(feedback.Feedback) == 0 {
		return bktID2FB, 0
	}
	total := 0
	sc := &stmtctx.StatementContext{TimeZone: time.UTC}
	min, max := GetMinValue(h.Tp), GetMaxValue(h.Tp)
	for _, fb := range feedback.Feedback {
		skip, err := fb.adjustFeedbackBoundaries(sc, &min, &max)
		if err != nil {
			logutil.Logger(context.Background()).Debug("adjust feedback boundaries failed", zap.Error(err))
			continue
		}
		if skip {
			continue
		}
		idx, _ := h.Bounds.LowerBound(0, fb.Lower)
		bktIdx := 0
		// The last bucket also stores the feedback that falls outside the upper bound.
		if idx >= h.Bounds.NumRows()-2 {
			bktIdx = h.Len() - 1
		} else {
			bktIdx = idx / 2
			// Make sure that this feedback lies within the bucket.
			if chunk.Compare(h.Bounds.GetRow(2*bktIdx+1), 0, fb.Upper) < 0 {
				continue
			}
		}
		total++
		bkt := bktID2FB[bktIdx]
		if bkt == nil {
			bkt = &BucketFeedback{lower: h.GetLower(bktIdx), upper: h.GetUpper(bktIdx)}
			bktID2FB[bktIdx] = bkt
		}
		bkt.feedback = append(bkt.feedback, fb)
		// Update the bound if necessary.
		res, err := bkt.lower.CompareDatum(nil, fb.Lower)
		if err != nil {
			logutil.Logger(context.Background()).Debug("compare datum failed", zap.Any("value1", bkt.lower), zap.Any("value2", fb.Lower), zap.Error(err))
			continue
		}
		if res > 0 {
			bkt.lower = fb.Lower
		}
		res, err = bkt.upper.CompareDatum(nil, fb.Upper)
		if err != nil {
			logutil.Logger(context.Background()).Debug("compare datum failed", zap.Any("value1", bkt.upper), zap.Any("value2", fb.Upper), zap.Error(err))
			continue
		}
		if res < 0 {
			bkt.upper = fb.Upper
		}
	}
	return bktID2FB, total
}
878736
func NewBucketService(s *platform.Source) (platform.BucketService, error) {
	switch s.Type {
	case platform.SelfSourceType:
		// TODO(fntlnz): this is supposed to call a bucket service directly locally,
		// we are letting it err for now since we have some refactoring to do on
		// how services are instantiated
		return nil, fmt.Errorf("self source type not implemented")
	case platform.V2SourceType:
		return &http.BucketService{
			Addr:               s.URL,
			InsecureSkipVerify: s.InsecureSkipVerify,
			Token:              s.Token,
		}, nil
	case platform.V1SourceType:
		return &influxdb.BucketService{
			Source: s,
		}, nil
	}
	return nil, fmt.Errorf("unsupported source type %s", s.Type)
}
303653
func initTowerDBBuckets(tx *bbolt.Tx) error {
	buckets := [][]byte{
		sessionsBkt,
		updateIndexBkt,
		updatesBkt,
		lookoutTipBkt,
	}

	for _, bucket := range buckets {
		_, err := tx.CreateBucketIfNotExists(bucket)
		if err != nil {
			return err
		}
	}

	return nil
}
612040
func (s *UpdateFindingsFeedbackInput) SetFeedback(v string) *UpdateFindingsFeedbackInput {
	s.Feedback = &v
	return s
}
630700
func createBuckets(ns walletdb.ReadWriteBucket) error {
	if _, err := ns.CreateBucket(bucketBlocks); err != nil {
		str := "failed to create blocks bucket"
		return storeError(ErrDatabase, str, err)
	}
	if _, err := ns.CreateBucket(bucketTxRecords); err != nil {
		str := "failed to create tx records bucket"
		return storeError(ErrDatabase, str, err)
	}
	if _, err := ns.CreateBucket(bucketCredits); err != nil {
		str := "failed to create credits bucket"
		return storeError(ErrDatabase, str, err)
	}
	if _, err := ns.CreateBucket(bucketDebits); err != nil {
		str := "failed to create debits bucket"
		return storeError(ErrDatabase, str, err)
	}
	if _, err := ns.CreateBucket(bucketUnspent); err != nil {
		str := "failed to create unspent bucket"
		return storeError(ErrDatabase, str, err)
	}
	if _, err := ns.CreateBucket(bucketUnmined); err != nil {
		str := "failed to create unmined bucket"
		return storeError(ErrDatabase, str, err)
	}
	if _, err := ns.CreateBucket(bucketUnminedCredits); err != nil {
		str := "failed to create unmined credits bucket"
		return storeError(ErrDatabase, str, err)
	}
	if _, err := ns.CreateBucket(bucketUnminedInputs); err != nil {
		str := "failed to create unmined inputs bucket"
		return storeError(ErrDatabase, str, err)
	}

	return nil
}
784598
func (s *StreamingLoggingConfig) SetBucket(v string) *StreamingLoggingConfig {
	s.Bucket = &v
	return s
}
824070
func (builder *StoreValueCommandBuilder) WithBucket(bucket string) *StoreValueCommandBuilder {
	builder.protobuf.Bucket = []byte(bucket)
	return builder
}
1149956
func (c *GetBuildersCall) Bucket(bucket ...string) *GetBuildersCall {
	c.urlParams_.SetMulti("bucket", append([]string{}, bucket...))
	return c
}
784423
func NewBucketService() *BucketService {
	return &BucketService{
		OpenFn:           func() error { return nil },
		CloseFn:          func() error { return nil },
		WithLoggerFn:     func(l *zap.Logger) {},
		FindBucketByIDFn: func(context.Context, platform.ID) (*platform.Bucket, error) { return nil, nil },
		FindBucketFn:     func(context.Context, platform.BucketFilter) (*platform.Bucket, error) { return nil, nil },
		FindBucketsFn: func(context.Context, platform.BucketFilter, ...platform.FindOptions) ([]*platform.Bucket, int, error) {
			return nil, 0, nil
		},
		CreateBucketFn: func(context.Context, *platform.Bucket) error { return nil },
		UpdateBucketFn: func(context.Context, platform.ID, platform.BucketUpdate) (*platform.Bucket, error) { return nil, nil },
		DeleteBucketFn: func(context.Context, platform.ID) error { return nil },
	}
}
696231
func (bm *bucketMeta) createBucket(name []byte) *bucketMeta {
	bm.bucketsLock.Lock()
	defer bm.bucketsLock.Unlock()

	// Always create a new Bucket since CreateBucket above fails if the
	// bucket already exists.
	b := newBucketMeta()
	bm.buckets[string(name)] = b
	return b
}
356914
func newBucket(tx *Tx) Bucket {
	var b = Bucket{tx: tx, FillPercent: DefaultFillPercent}
	if tx.writable {
		b.buckets = make(map[string]*Bucket)
		b.nodes = make(map[pgid]*node)
	}
	return b
}
27339
func (s *BucketService) UpdateBucket(ctx context.Context, id influxdb.ID, upd influxdb.BucketUpdate) (*influxdb.Bucket, error) {
	b, err := s.s.FindBucketByID(ctx, id)
	if err != nil {
		return nil, err
	}

	if err := authorizeWriteBucket(ctx, b.OrgID, id); err != nil {
		return nil, err
	}

	return s.s.UpdateBucket(ctx, id, upd)
}
318527
func (s *JobReport) SetBucket(v string) *JobReport {
	s.Bucket = &v
	return s
}
361800
func (bk *Backend) pathToBucket(bucket string) string {
	return filepath.Join(bk.rootDir, bucket)
}
766545
func newBucket(cap int, period time.Duration) (b *Bucket) {
	b = &Bucket{
		cap:         cap,
		channelQuit: make(chan bool),
		count:       0,
		period:      period,
	}
	go b.tick()
	return
}
530487
func (builder *UpdateCounterCommandBuilder) WithBucket(bucket string) *UpdateCounterCommandBuilder {
	builder.bucket = bucket
	return builder
}
748242
func (b *Bucket) CreateBucketIfNotExists(name []byte) (*Bucket, error) {
	bb, err := b.boltBucket.CreateBucketIfNotExists(name)
	if err != nil {
		return nil, err
	}

	bmeta := b.bm.getOrCreateBucket(name)
	return newBucket(bmeta, bb), nil
}