streaming buffer fix

rubywtl · rubywtl · commit 4542e0be07e8 · 2025-08-22T09:17:28.000-07:00
diff --git a/pkg/distributed_execution/remote_node.go b/pkg/distributed_execution/remote_node.go
@@ -163,7 +163,7 @@ func newDistributedRemoteExecution(ctx context.Context, pool *client.Pool, fragm
 		batchSize:   1000,
 		fragmentKey: fragmentKey,
 		addr:        childIDToAddr[fragmentKey.fragmentID],
-		buffer:      nil,
+		buffer:      []model.StepVector{},
 		bufferIndex: 0,
 		initialized: false,
 	}
@@ -288,7 +288,7 @@ func (d *DistributedRemoteExecution) Next(ctx context.Context) ([]model.StepVect
 		}
 	}
 
-	end := int(d.batchSize)
+	end := len(d.buffer)
 	if end > len(d.buffer) {
 		end = len(d.buffer)
 	}

Original file line number	Diff line number	Diff line change
`@@ -163,7 +163,7 @@ func newDistributedRemoteExecution(ctx context.Context, pool *client.Pool, fragm`
`163`	`163`	`batchSize: 1000,`
`164`	`164`	`fragmentKey: fragmentKey,`
`165`	`165`	`addr: childIDToAddr[fragmentKey.fragmentID],`
`166`		`- buffer: nil,`
	`166`	`+ buffer: []model.StepVector{},`
`167`	`167`	`bufferIndex: 0,`
`168`	`168`	`initialized: false,`
`169`	`169`	`}`
`@@ -288,7 +288,7 @@ func (d *DistributedRemoteExecution) Next(ctx context.Context) ([]model.StepVect`
`288`	`288`	`}`
`289`	`289`	`}`
`290`	`290`
`291`		`- end := int(d.batchSize)`
	`291`	`+ end := len(d.buffer)`
`292`	`292`	`if end > len(d.buffer) {`
`293`	`293`	`end = len(d.buffer)`
`294`	`294`	`}`