func (s *stageContextImpl) SetReductionOperation(val sif.ReductionOperation) error {
	if s.ReductionOperation() != nil {
		return fmt.Errorf("cannot overwrite ReductionOperation for Stage (already set)")
	}
	s.state.reduceFn = val
	return nil
}
Configure the reduction operation for the end of this stage
package cluster
import (
	"context"
	"fmt"
	"time"
	"github.com/go-sif/sif"
	itypes "github.com/go-sif/sif/internal/types"
)
const stageContextStateKey = "sif.cluster.stageContextState"
type stageContextState struct {
	stage               itypes.Stage
	outgoingSchema      sif.Schema
	shuffleBuckets      []uint64
	pCache              sif.PartitionCache
	pIndex              sif.PartitionIndex
	pIncoming           sif.PartitionIterator
	keyFn               sif.KeyingOperation
	reduceFn            sif.ReductionOperation
	accumulator         sif.Accumulator
	collectionLimit     int
	targetPartitionSize int
}
type stageContextImpl struct {
	ctx   context.Context
	state *stageContextState
}
func createStageContext(ctx context.Context, stage itypes.Stage) sif.StageContext {
	var state *stageContextState
	if i := ctx.Value(stageContextStateKey); i != nil {
		state = i.(*stageContextState)
	} else {
		state = &stageContextState{
			stage:               stage,
			collectionLimit:     -1,
			targetPartitionSize: -1,
		}
	}
	return &stageContextImpl{
		ctx:   ctx,
		state: state,
	}
}
func (s *stageContextImpl) Deadline() (deadline time.Time, ok bool) {
	return s.ctx.Deadline()
}
func (s *stageContextImpl) Done() <-chan struct{} {
	return s.ctx.Done()
}
func (s *stageContextImpl) Err() error {
	return s.ctx.Err()
}
func (s *stageContextImpl) Value(key interface{}) interface{} {
	return s.ctx.Value(key)
}
func (s *stageContextImpl) OutgoingSchema() sif.Schema {
	return s.state.outgoingSchema
}
func (s *stageContextImpl) SetOutgoingSchema(schema sif.Schema) error {
	if s.OutgoingSchema() != nil {
		return fmt.Errorf("cannot overwrite outgoing Schema for current Stage (already set)")
	}
	s.state.outgoingSchema = schema
	return nil
}
func (s *stageContextImpl) ShuffleBuckets() []uint64 {
	return s.state.shuffleBuckets
}
func (s *stageContextImpl) SetShuffleBuckets(buckets []uint64) error {
	if s.ShuffleBuckets() != nil {
		return fmt.Errorf("cannot overwrite shuffle buckets for Stage (already set)")
	}
	s.state.shuffleBuckets = buckets
	return nil
}
func (s *stageContextImpl) PartitionCache() sif.PartitionCache {
	return s.state.pCache
}
func (s *stageContextImpl) SetPartitionCache(cache sif.PartitionCache) error {
	if s.PartitionCache() != nil {
		return fmt.Errorf("cannot overwrite PartitionCache for Stage (already set)")
	}
	s.state.pCache = cache
	return nil
}
func (s *stageContextImpl) PartitionIndex() sif.PartitionIndex {
	return s.state.pIndex
}
func (s *stageContextImpl) SetPartitionIndex(idx sif.PartitionIndex) error {
	if s.PartitionIndex() != nil {
		return fmt.Errorf("cannot overwrite PartitionIndex for Stage (already set)")
	}
	s.state.pIndex = idx
	return nil
}
func (s *stageContextImpl) IncomingPartitionIterator() sif.PartitionIterator {
	return s.state.pIncoming
}
func (s *stageContextImpl) SetIncomingPartitionIterator(i sif.PartitionIterator) error {
	if s.IncomingPartitionIterator() != nil {
		return fmt.Errorf("cannot overwrite IncomingPartitionIterator for Stage (already set)")
	}
	s.state.pIncoming = i
	return nil
}
func (s *stageContextImpl) KeyingOperation() sif.KeyingOperation {
	return s.state.keyFn
}
func (s *stageContextImpl) SetKeyingOperation(val sif.KeyingOperation) error {
	if s.KeyingOperation() != nil {
		return fmt.Errorf("cannot overwrite KeyingOperation for Stage (already set)")
	}
	s.state.keyFn = val
	return nil
}
func (s *stageContextImpl) ReductionOperation() sif.ReductionOperation {
	return s.state.reduceFn
}