func (pe *planExecutorImpl) GetShufflePartitionIterator(bucket uint64) (sif.SerializedPartitionIterator, error) {
	var idx sif.PartitionIndex
	pe.shuffleLock.Lock()
	defer pe.shuffleLock.Unlock()
	if pe.shuffleReady == nil {
		return nil, fmt.Errorf("no PartitionIndex available for shuffle")
	}
	bpi, ok := pe.shuffleReady.(sif.BucketedPartitionIndex)
	if ok {
		idx = bpi.GetBucket(bucket)
		if idx == nil {
			return nil, fmt.Errorf("no PartitionIterator available for bucket %d", bucket)
		}
	} else {
		idx = pe.shuffleReady
	}
	return idx.GetSerializedPartitionIterator(true), nil
}
GetShufflePartitionIterator serves up an iterator for partitions to shuffle
package dataframe
import (
	"context"
	"fmt"
	"log"
	"runtime"
	"sync"
	"time"
	"github.com/go-sif/sif"
	"github.com/go-sif/sif/errors"
	"github.com/go-sif/sif/internal/partition"
	"github.com/go-sif/sif/internal/pcache"
	pb "github.com/go-sif/sif/internal/rpc"
	"github.com/go-sif/sif/internal/stats"
	itypes "github.com/go-sif/sif/internal/types"
	uuid "github.com/gofrs/uuid"
	"github.com/hashicorp/go-multierror"
)
type planExecutorImpl struct {
	id   string
	plan itypes.Plan
	conf *itypes.PlanExecutorConfig
	done context.CancelFunc
	nextStage            int
	partitionLoaders     []sif.PartitionLoader
	partitionLoadersLock sync.Mutex
	assignedBucket       uint64
	shuffleIterators     map[uint64]sif.SerializedPartitionIterator  
	shuffleIteratorsLock sync.Mutex
	shuffleLock          sync.Mutex
	shuffleReady         sif.PartitionIndex
	accumulateReady      sif.Accumulator
	statsTracker         *stats.RunStatistics
	pCache               sif.PartitionCache
}
func CreatePlanExecutor(ctx context.Context, plan itypes.Plan, conf *itypes.PlanExecutorConfig, statsTracker *stats.RunStatistics, isCoordinator bool) itypes.PlanExecutor {
	id, err := uuid.NewV4()
	if err != nil {
		log.Fatalf("failed to generate UUID: %v", err)
	}
	executor := &planExecutorImpl{
		id:               id.String(),
		plan:             plan,
		conf:             conf,
		done:             func() {},
		partitionLoaders: make([]sif.PartitionLoader, 0),
		shuffleIterators: make(map[uint64]sif.SerializedPartitionIterator),
		statsTracker:     statsTracker,
	}
	if conf.CacheMemoryHighWatermark == 0 {
		conf.CacheMemoryHighWatermark = 512 * 1024 * 1024  
	}
	if conf.CacheMemoryInitialSize == 0 {
		conf.CacheMemoryInitialSize = 32 * 1024  
	}
	executor.pCache = pcache.NewLRU(&pcache.LRUConfig{
		InitialSize: conf.CacheMemoryInitialSize,
		DiskPath:    conf.TempFilePath,
		Compressor:  conf.Compressor,
	})
	if !isCoordinator {
		monitorCtx, canceller := context.WithCancel(ctx)
		executor.done = canceller
		go executor.monitorMemoryUsage(monitorCtx)
	}
	return executor
}
func (pe *planExecutorImpl) ID() string {
	return pe.id
}
func (pe *planExecutorImpl) GetConf() *itypes.PlanExecutorConfig {
	return pe.conf
}
func (pe *planExecutorImpl) Stop() {
	pe.done()
	pe.shuffleLock.Lock()
	defer pe.shuffleLock.Unlock()
	if pe.pCache != nil {
		pe.pCache.Destroy()
	}
}
func (pe *planExecutorImpl) HasNextStage() bool {
	return pe.nextStage < (pe.plan.Size())
}
func (pe *planExecutorImpl) GetNextStage() itypes.Stage {
	if pe.nextStage >= pe.plan.Size() {
		return nil
	}
	s := pe.plan.GetStage(pe.nextStage)
	pe.nextStage++
	if pe.conf.Streaming {
		pe.nextStage = pe.nextStage % pe.plan.Size()
	}
	return s
}
func (pe *planExecutorImpl) GetNumStages() int {
	return pe.plan.Size()
}
func (pe *planExecutorImpl) peekNextStage() itypes.Stage {
	if pe.HasNextStage() {
		return pe.plan.GetStage(pe.nextStage)
	}
	return nil
}
func (pe *planExecutorImpl) GetCurrentStage() itypes.Stage {
	if pe.nextStage == 0 {
		return pe.plan.GetStage(pe.nextStage)
	}
	return pe.plan.GetStage(pe.nextStage - 1)
}
func (pe *planExecutorImpl) onFirstStage() bool {
	return pe.nextStage == 1
}
func (pe *planExecutorImpl) hasPartitionLoaders() bool {
	pe.partitionLoadersLock.Lock()
	defer pe.partitionLoadersLock.Unlock()
	return len(pe.partitionLoaders) > 0
}
func (pe *planExecutorImpl) IsShuffleReady() bool {
	return pe.shuffleReady != nil
}
func (pe *planExecutorImpl) IsAccumulatorReady() bool {
	return pe.accumulateReady != nil
}
func (pe *planExecutorImpl) AssignPartitionLoader(sLoader []byte) error {
	loader, err := pe.plan.Source().DeserializeLoader(sLoader[0:])
	if err != nil {
		return err
	}
	pe.partitionLoadersLock.Lock()
	defer pe.partitionLoadersLock.Unlock()
	pe.partitionLoaders = append(pe.partitionLoaders, loader)
	return nil
}
func (pe *planExecutorImpl) InitStageContext(sctx sif.StageContext, stage itypes.Stage) error {
	if pe.plan.Size() == 0 {
		return fmt.Errorf("plan has no stages")
	}
	nextStage := pe.peekNextStage()
	if nextStage == nil {
		nextStage = pe.GetCurrentStage()
	}
	sctx.SetOutgoingSchema(nextStage.IncomingSchema())
	sctx.SetPartitionCache(pe.pCache)
	if pe.onFirstStage() && pe.hasPartitionLoaders() {
		it := createPartitionLoaderIterator(pe.partitionLoaders, pe.plan.Parser())
		if !pe.conf.Streaming {
			pe.partitionLoadersLock.Lock()
			pe.partitionLoaders = make([]sif.PartitionLoader, 0)  
			pe.partitionLoadersLock.Unlock()
		}
		if err := sctx.SetIncomingPartitionIterator(it); err != nil {
			return err
		}
	} else if pe.shuffleReady != nil {
		pe.shuffleLock.Lock()
		defer pe.shuffleLock.Unlock()
		bpi, ok := pe.shuffleReady.(sif.BucketedPartitionIndex)
		if !ok {
			return fmt.Errorf("partitionIndex available as source for next Stage is not a BucketedPartitionIndex")
		}
		idx := bpi.GetBucket(pe.assignedBucket)
		it := createPreloadingPartitionIterator(idx.GetPartitionIterator(true), 3)
		if err := sctx.SetIncomingPartitionIterator(it); err != nil {
			return err
		}
		pe.shuffleReady = nil
		pe.accumulateReady = nil
	}
	err := stage.WorkerInitialize(sctx)
	if err != nil {
		return err
	}
	return nil
}
func (pe *planExecutorImpl) TransformPartitions(sctx sif.StageContext, fn itypes.PartitionTransform, onRowError func(error) error) error {
	parts := sctx.IncomingPartitionIterator()
	currentStageID := pe.GetCurrentStage().ID()
	for parts.HasNextPartition() {
		part, unlockPartition, err := parts.NextPartition()
		if _, ok := err.(errors.NoMorePartitionsError); ok {
			if unlockPartition != nil {
				unlockPartition()
			}
			break
		} else if err != nil {
			if unlockPartition != nil {
				unlockPartition()
			}
			return err
		}
		pe.statsTracker.StartPartition()
		opart := part.(sif.OperablePartition)
		_, err = fn(sctx, opart)  
		if err := onRowError(err); err != nil {
			if unlockPartition != nil {
				unlockPartition()
			}
			return err
		}
		if unlockPartition != nil {
			unlockPartition()
		}
		pe.statsTracker.EndPartition(currentStageID, part.GetNumRows())
	}
	pe.shuffleLock.Lock()
	defer pe.shuffleLock.Unlock()
	if sctx.PartitionIndex() != nil && sctx.PartitionIndex().NumPartitions() > 0 {
		pe.shuffleReady = sctx.PartitionIndex()
	} else if sctx.Accumulator() != nil {
		pe.accumulateReady = sctx.Accumulator()
	}
	return nil
}
func (pe *planExecutorImpl) AssignShuffleBucket(assignedBucket uint64) {
	pe.assignedBucket = assignedBucket
}