partial (unsquashed) stores no longer contain traceID in filename, so…

… they can be reused, squasher will handle this gracefully

the execouts for stores are now only written on a PartialKV, containing
a list of operations to re-apply. They are only output as deltas when
applied to a fullKV.

docs/release-notes/change-log.md

@@ -22,7 +22,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 * Tier2 will now read back mapper outputs (if they exist) to prevent running them again. Additionally, it will not read back the full blocks if its inputs can be satisfied from existing cached mapper outputs.
 
-* Tier2 will skip processing completely if the `output_module` is a mapper that has already been processed (ex: when multiple requests are indexing the same data at the same time)
+* Tier2 will skip processing completely if it's processing the last stage and the `output_module` is a mapper that has already been processed (ex: when multiple requests are indexing the same data at the same time)
+
+* Tier2 will skip processing completely if it's processing a stage that is not the last, but all the stores and outputs have been processed and cached.
+
+* The "partial" store outputs no longer contain the trace ID in the filename, allowing them to be reused. If many requests point to the same modules being squashed, the squasher will detect if another Tier1 has squashed its file and reload the store from the produced full KV.
 
 * [Operator] Readiness metric for Substreams tier1 app is now named `substreams_tier1` (was mistakenly called `firehose` before).
 

orchestrator/parallelprocessor.go

@@ -33,13 +33,12 @@ func BuildParallelProcessor(
 	execoutStorage *execout.Configs,
 	respFunc func(resp substreams.ResponseFromAnyTier) error,
 	storeConfigs store.ConfigMap,
-	traceID string,
 ) (*ParallelProcessor, error) {
 
 	stream := response.New(respFunc)
 	sched := scheduler.New(ctx, stream)
 
-	stages := stage.NewStages(ctx, outputGraph, reqPlan, storeConfigs, traceID)
+	stages := stage.NewStages(ctx, outputGraph, reqPlan, storeConfigs)
 	sched.Stages = stages
 
 	// OPTIMIZATION: We should fetch the ExecOut files too, and see if they
@@ -91,7 +90,6 @@ func BuildParallelProcessor(
 			reqPlan.StoresSegmenter(),
 			storeConfigs,
 			execoutStorage,
-			traceID,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("fetch stores storage state: %w", err)
@@ -102,7 +100,6 @@ func BuildParallelProcessor(
 			reqPlan.WriteOutSegmenter(),
 			storeConfigs,
 			execoutStorage,
-			traceID,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("fetch stores storage state: %w", err)

orchestrator/stage/fetchstorage.go

@@ -17,7 +17,6 @@ func (s *Stages) FetchStoresState(
 	segmenter *block.Segmenter,
 	storeConfigMap store.ConfigMap,
 	execoutConfigs *execout.Configs,
-	traceID string,
 ) error {
 	completes := make(unitMap)
 	partials := make(unitMap)
@@ -103,9 +102,6 @@ func (s *Stages) FetchStoresState(
 				if !rng.Equals(partial.Range) {
 					continue
 				}
-				if traceID != partial.TraceID {
-					continue
-				}
 				unit := Unit{Stage: stageIdx, Segment: segmentIdx}
 
 				if s.getState(unit) == UnitCompleted {

orchestrator/stage/squash.go

@@ -58,7 +58,7 @@ func (s *Stages) singleSquash(stage *Stage, modState *StoreModuleState, mergeUni
 	metrics.start = time.Now()
 
 	rng := modState.segmenter.Range(mergeUnit.Segment)
-	partialFile := store.NewPartialFileInfo(modState.name, rng.StartBlock, rng.ExclusiveEndBlock, s.traceID)
+	partialFile := store.NewPartialFileInfo(modState.name, rng.StartBlock, rng.ExclusiveEndBlock)
 	partialKV := modState.derivePartialKV(rng.StartBlock)
 	segmentEndsOnInterval := modState.segmenter.EndsOnInterval(mergeUnit.Segment)
 
@@ -72,6 +72,12 @@ func (s *Stages) singleSquash(stage *Stage, modState *StoreModuleState, mergeUni
 	// Load
 	metrics.loadStart = time.Now()
 	if err := partialKV.Load(s.ctx, partialFile); err != nil {
+		if nextFull, err := modState.getStore(s.ctx, rng.ExclusiveEndBlock); err == nil { // try to load an already-merged file
+			s.logger.Info("got full store from cache instead of partial, reloading", zap.Stringer("store", nextFull))
+			modState.cachedStore = nextFull
+			modState.lastBlockInStore = rng.ExclusiveEndBlock
+			return nil
+		}
 		return fmt.Errorf("loading partial: %q: %w", partialFile.Filename, err)
 	}
 	metrics.loadEnd = time.Now()

orchestrator/stage/stages.go

@@ -32,9 +32,8 @@ import (
 // that the Stage is completed, kicking off the next layer of jobs.
 
 type Stages struct {
-	ctx     context.Context
-	logger  *zap.Logger
-	traceID string
+	ctx    context.Context
+	logger *zap.Logger
 
 	globalSegmenter *block.Segmenter // This segmenter covers both the stores and the mapper
 	storeSegmenter  *block.Segmenter // This segmenter covers only jobs needed to build up stores according to the RequestPlan.
@@ -58,7 +57,6 @@ func NewStages(
 	outputGraph *outputmodules.Graph,
 	reqPlan *plan.RequestPlan,
 	storeConfigs store.ConfigMap,
-	traceID string,
 ) (out *Stages) {
 
 	if !reqPlan.RequiresParallelProcessing() {
@@ -70,7 +68,6 @@ func NewStages(
 	stagedModules := outputGraph.StagedUsedModules()
 	out = &Stages{
 		ctx:             ctx,
-		traceID:         traceID,
 		logger:          reqctx.Logger(ctx),
 		globalSegmenter: reqPlan.BackprocessSegmenter(),
 	}

orchestrator/stage/stages_test.go

@@ -23,7 +23,6 @@ func TestNewStages(t *testing.T) {
 		outputmodules.TestGraphStagedModules(5, 7, 12, 22, 25),
 		reqPlan,
 		nil,
-		"trace",
 	)
 
 	assert.Equal(t, 8, stages.globalSegmenter.Count()) // from 5 to 75
@@ -49,7 +48,6 @@ func TestNewStagesNextJobs(t *testing.T) {
 		outputmodules.TestGraphStagedModules(5, 5, 5, 5, 5),
 		reqPlan,
 		nil,
-		"trace",
 	)
 
 	stages.allocSegments(0)

orchestrator/work/worker.go

@@ -288,7 +288,7 @@ func toRPCPartialFiles(completed *pbssinternal.Completed) (out store.FileInfos)
 	// stores to all having been processed.
 	out = make(store.FileInfos, len(completed.AllProcessedRanges))
 	for i, b := range completed.AllProcessedRanges {
-		out[i] = store.NewPartialFileInfo("TODO:CHANGE-ME", b.StartBlock, b.EndBlock, completed.TraceId)
+		out[i] = store.NewPartialFileInfo("TODO:CHANGE-ME", b.StartBlock, b.EndBlock)
 	}
 	return
 }

pb/last_generate.txt

@@ -1,2 +1,2 @@
-generate.sh - Thu 14 Mar 2024 10:02:19 EDT - arnaudberger
-streamingfast/proto revision: 2999cd42d71a82c4adf739557f9520f0609f7b10
+generate.sh - Tue Mar 19 09:09:54 EDT 2024 - stepd
+streamingfast/proto revision: fde3637cef38103a68301d0ca19b3f2af9b6079b