engine

type Config ¶

type Config struct {
	// MaxBundleSize caps the number of elements permitted in a bundle.
	// 0 or less means this is ignored.
	MaxBundleSize int
}

type ElementManager ¶

type ElementManager struct {
	// contains filtered or unexported fields
}

There comes a time in every Beam runner's life that they develop a type called a Watermark Manager, that becomes the core of the runner. Given that handling the watermark is the core of the Beam model, this is not super surprising.

Essentially, it needs to track the current watermarks for each pcollection and transform/stage. But it's tricky, since the watermarks for the PCollections are always relative to transforms/stages.

Key parts:

• The parallel input's PCollection's watermark is relative to committed consumed elements. That is, the input elements consumed by the transform after a successful bundle, can advance the watermark, based on the minimum of their elements.
• An output PCollection's watermark is relative to its producing transform, which relates to *all of it's outputs*.

This means that a PCollection's watermark is the minimum of all it's consuming transforms.

So, the watermark manager needs to track: Pending Elements for each stage, along with their windows and timestamps. Each transform's view of the watermarks for the PCollections.

Watermarks are always advanced based on consumed input.

func NewElementManager(config Config) *ElementManager

func (em *ElementManager) AddStage(ID string, inputIDs, sides, outputIDs []string)

func (em *ElementManager) Bundles(ctx context.Context, nextBundID func() string) <-chan RunBundle

func (em *ElementManager) Impulse(stageID string)

func (em *ElementManager) InputForBundle(rb RunBundle, info PColInfo) [][]byte

func (em *ElementManager) PersistBundle(rb RunBundle, col2Coders map[string]PColInfo, d TentativeData, inputInfo PColInfo, residuals [][]byte)

type PColInfo ¶

type PColInfo struct {
	GlobalID string
	WDec     exec.WindowDecoder
	WEnc     exec.WindowEncoder
	EDec     func(io.Reader) []byte
}

type RunBundle ¶

type RunBundle struct {
	StageID   string
	BundleID  string
	Watermark mtime.Time
}

func (rb RunBundle) LogValue() slog.Value

type TentativeData ¶

type TentativeData struct {
	Raw map[string][][]byte
}

func (d *TentativeData) WriteData(colID string, data []byte)